Merge branch 'main' into codex/add-session-loading-logic-in-codex-rs

now storing the sessions in `~/.codex/sessions/YYYY/MM/DD/<file>`

codex-cli/bin/codex.js

@@ -15,7 +15,6 @@
  *      current platform / architecture, an error is thrown.
  */
 
-import { spawnSync } from "child_process";
 import fs from "fs";
 import path from "path";
 import { fileURLToPath, pathToFileURL } from "url";
@@ -35,7 +34,7 @@ const wantsNative = fs.existsSync(path.join(__dirname, "use-native")) ||
     : false);
 
 // Try native binary if requested.
-if (wantsNative) {
+if (wantsNative && process.platform !== 'win32') {
   const { platform, arch } = process;
 
   let targetTriple = null;
@@ -74,22 +73,76 @@ if (wantsNative) {
   }
 
   const binaryPath = path.join(__dirname, "..", "bin", `codex-${targetTriple}`);
-  const result = spawnSync(binaryPath, process.argv.slice(2), {
+
+  // Use an asynchronous spawn instead of spawnSync so that Node is able to
+  // respond to signals (e.g. Ctrl-C / SIGINT) while the native binary is
+  // executing. This allows us to forward those signals to the child process
+  // and guarantees that when either the child terminates or the parent
+  // receives a fatal signal, both processes exit in a predictable manner.
+  const { spawn } = await import("child_process");
+
+  const child = spawn(binaryPath, process.argv.slice(2), {
     stdio: "inherit",
   });
 
-  const exitCode = typeof result.status === "number" ? result.status : 1;
-  process.exit(exitCode);
-}
+  child.on("error", (err) => {
+    // Typically triggered when the binary is missing or not executable.
+    // Re-throwing here will terminate the parent with a non-zero exit code
+    // while still printing a helpful stack trace.
+    // eslint-disable-next-line no-console
+    console.error(err);
+    process.exit(1);
+  });
 
-// Fallback: execute the original JavaScript CLI.
+  // Forward common termination signals to the child so that it shuts down
+  // gracefully. In the handler we temporarily disable the default behavior of
+  // exiting immediately; once the child has been signaled we simply wait for
+  // its exit event which will in turn terminate the parent (see below).
+  const forwardSignal = (signal) => {
+    if (child.killed) {
+      return;
+    }
+    try {
+      child.kill(signal);
+    } catch {
+      /* ignore */
+    }
+  };
 
-// Resolve the path to the compiled CLI bundle
-const cliPath = path.resolve(__dirname, "../dist/cli.js");
-const cliUrl = pathToFileURL(cliPath).href;
+  ["SIGINT", "SIGTERM", "SIGHUP"].forEach((sig) => {
+    process.on(sig, () => forwardSignal(sig));
+  });
 
-// Load and execute the CLI
-(async () => {
+  // When the child exits, mirror its termination reason in the parent so that
+  // shell scripts and other tooling observe the correct exit status.
+  // Wrap the lifetime of the child process in a Promise so that we can await
+  // its termination in a structured way. The Promise resolves with an object
+  // describing how the child exited: either via exit code or due to a signal.
+  const childResult = await new Promise((resolve) => {
+    child.on("exit", (code, signal) => {
+      if (signal) {
+        resolve({ type: "signal", signal });
+      } else {
+        resolve({ type: "code", exitCode: code ?? 1 });
+      }
+    });
+  });
+
+  if (childResult.type === "signal") {
+    // Re-emit the same signal so that the parent terminates with the expected
+    // semantics (this also sets the correct exit code of 128 + n).
+    process.kill(process.pid, childResult.signal);
+  } else {
+    process.exit(childResult.exitCode);
+  }
+} else {
+  // Fallback: execute the original JavaScript CLI.
+
+  // Resolve the path to the compiled CLI bundle
+  const cliPath = path.resolve(__dirname, "../dist/cli.js");
+  const cliUrl = pathToFileURL(cliPath).href;
+
+  // Load and execute the CLI
   try {
     await import(cliUrl);
   } catch (err) {
@@ -97,4 +150,4 @@ const cliUrl = pathToFileURL(cliPath).href;
     console.error(err);
     process.exit(1);
   }
-})();
+}

codex-rs/Cargo.lock

@@ -683,6 +683,7 @@ dependencies = [
  "tree-sitter",
  "tree-sitter-bash",
  "uuid",
+ "walkdir",
  "wildmatch",
  "wiremock",
 ]

codex-rs/core/Cargo.toml

@@ -65,4 +65,5 @@ predicates = "3"
 pretty_assertions = "1.4.1"
 tempfile = "3"
 tokio-test = "0.4"
+walkdir = "2.5.0"
 wiremock = "0.6"

codex-rs/core/src/chat_completions.rs

@@ -134,7 +134,7 @@ pub(crate) async fn stream_chat_completions(
 
         match res {
             Ok(resp) if resp.status().is_success() => {
-                let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(16);
+                let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
                 let stream = resp.bytes_stream().map_err(CodexErr::Reqwest);
                 tokio::spawn(process_chat_sse(stream, tx_event));
                 return Ok(ResponseStream { rx_event });
@@ -426,6 +426,12 @@ where
                     // will never appear in a Chat Completions stream.
                     continue;
                 }
+                Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(_))))
+                | Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(_)))) => {
+                    // Deltas are ignored here since aggregation waits for the
+                    // final OutputItemDone.
+                    continue;
+                }
             }
         }
     }

codex-rs/core/src/client.rs

@@ -125,6 +125,7 @@ impl ModelClient {
             reasoning,
             previous_response_id: prompt.prev_id.clone(),
             store: prompt.store,
+            // TODO: make this configurable
             stream: true,
         };
 
@@ -148,7 +149,7 @@ impl ModelClient {
             let res = req_builder.send().await;
             match res {
                 Ok(resp) if resp.status().is_success() => {
-                    let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(16);
+                    let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
 
                     // spawn task to process SSE
                     let stream = resp.bytes_stream().map_err(CodexErr::Reqwest);
@@ -205,6 +206,7 @@ struct SseEvent {
     kind: String,
     response: Option<Value>,
     item: Option<Value>,
+    delta: Option<String>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -337,6 +339,22 @@ where
                     return;
                 }
             }
+            "response.output_text.delta" => {
+                if let Some(delta) = event.delta {
+                    let event = ResponseEvent::OutputTextDelta(delta);
+                    if tx_event.send(Ok(event)).await.is_err() {
+                        return;
+                    }
+                }
+            }
+            "response.reasoning_summary_text.delta" => {
+                if let Some(delta) = event.delta {
+                    let event = ResponseEvent::ReasoningSummaryDelta(delta);
+                    if tx_event.send(Ok(event)).await.is_err() {
+                        return;
+                    }
+                }
+            }
             "response.created" => {
                 if event.response.is_some() {
                     let _ = tx_event.send(Ok(ResponseEvent::Created {})).await;
@@ -360,10 +378,8 @@ where
             | "response.function_call_arguments.delta"
             | "response.in_progress"
             | "response.output_item.added"
-            | "response.output_text.delta"
             | "response.output_text.done"
             | "response.reasoning_summary_part.added"
-            | "response.reasoning_summary_text.delta"
             | "response.reasoning_summary_text.done" => {
                 // Currently, we ignore these events, but we handle them
                 // separately to skip the logging message in the `other` case.
@@ -375,7 +391,7 @@ where
 
 /// used in tests to stream from a text SSE file
 async fn stream_from_fixture(path: impl AsRef<Path>) -> Result<ResponseStream> {
-    let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(16);
+    let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
     let f = std::fs::File::open(path.as_ref())?;
     let lines = std::io::BufReader::new(f).lines();
 

codex-rs/core/src/client_common.rs

@@ -57,6 +57,8 @@ pub enum ResponseEvent {
         response_id: String,
         token_usage: Option<TokenUsage>,
     },
+    OutputTextDelta(String),
+    ReasoningSummaryDelta(String),
 }
 
 #[derive(Debug, Serialize)]

codex-rs/core/src/codex.rs

@@ -61,7 +61,9 @@ use crate::models::ResponseInputItem;
 use crate::models::ResponseItem;
 use crate::models::ShellToolCallParams;
 use crate::project_doc::get_user_instructions;
+use crate::protocol::AgentMessageDeltaEvent;
 use crate::protocol::AgentMessageEvent;
+use crate::protocol::AgentReasoningDeltaEvent;
 use crate::protocol::AgentReasoningEvent;
 use crate::protocol::ApplyPatchApprovalRequestEvent;
 use crate::protocol::AskForApproval;
@@ -103,7 +105,7 @@ impl Codex {
     /// submitted to start the session.
     pub async fn spawn(config: Config, ctrl_c: Arc<Notify>) -> CodexResult<(Codex, String)> {
         let (tx_sub, rx_sub) = async_channel::bounded(64);
-        let (tx_event, rx_event) = async_channel::bounded(64);
+        let (tx_event, rx_event) = async_channel::bounded(1600);
 
         let instructions = get_user_instructions(&config).await;
         let configure_session = Op::ConfigureSession {
@@ -200,6 +202,20 @@ impl Session {
             .map(PathBuf::from)
             .map_or_else(|| self.cwd.clone(), |p| self.cwd.join(p))
     }
+
+    pub async fn load_rollout(&self, path: std::path::PathBuf) -> std::io::Result<()> {
+        let (rec, saved) = crate::rollout::RolloutRecorder::resume(&path).await?;
+        {
+            let mut state = self.state.lock().unwrap();
+            state.previous_response_id = saved.state.previous_response_id;
+            if let Some(transcript) = state.zdr_transcript.as_mut() {
+                transcript.record_items(saved.items.iter());
+            }
+        }
+        let mut guard = self.rollout.lock().unwrap();
+        *guard = Some(rec);
+        Ok(())
+    }
 }
 
 /// Mutable state of the agent
@@ -307,6 +323,7 @@ impl Session {
     async fn record_conversation_items(&self, items: &[ResponseItem]) {
         debug!("Recording items for conversation: {items:?}");
         self.record_rollout_items(items).await;
+        self.record_state_snapshot().await;
 
         if let Some(transcript) = self.state.lock().unwrap().zdr_transcript.as_mut() {
             transcript.record_items(items);
@@ -330,6 +347,26 @@ impl Session {
         }
     }
 
+    async fn record_state_snapshot(&self) {
+        let snapshot = {
+            let state = self.state.lock().unwrap();
+            crate::rollout::SessionStateSnapshot {
+                previous_response_id: state.previous_response_id.clone(),
+            }
+        };
+
+        let recorder = {
+            let guard = self.rollout.lock().unwrap();
+            guard.as_ref().cloned()
+        };
+
+        if let Some(rec) = recorder {
+            if let Err(e) = rec.record_state(snapshot).await {
+                error!("failed to record rollout state: {e:#}");
+            }
+        }
+    }
+
     async fn notify_exec_command_begin(&self, sub_id: &str, call_id: &str, params: &ExecParams) {
         let event = Event {
             id: sub_id.to_string(),
@@ -742,6 +779,22 @@ async fn submission_loop(
                     other => sess.notify_approval(&id, other),
                 }
             }
+            Op::LoadSession { path } => {
+                let sess = match sess.as_ref() {
+                    Some(sess) => sess,
+                    None => {
+                        send_no_session_event(sub.id).await;
+                        continue;
+                    }
+                };
+                if let Err(e) = sess.load_rollout(path).await {
+                    let event = Event {
+                        id: sub.id,
+                        msg: EventMsg::Error(ErrorEvent { message: e.to_string() }),
+                    };
+                    tx_event.send(event).await.ok();
+                }
+            }
             Op::AddToHistory { text } => {
                 let id = session_id;
                 let config = config.clone();
@@ -1121,15 +1174,8 @@ async fn try_run_turn(
 
     let mut stream = sess.client.clone().stream(&prompt).await?;
 
-    // Buffer all the incoming messages from the stream first, then execute them.
-    // If we execute a function call in the middle of handling the stream, it can time out.
-    let mut input = Vec::new();
-    while let Some(event) = stream.next().await {
-        input.push(event?);
-    }
-
     let mut output = Vec::new();
-    for event in input {
+    while let Some(Ok(event)) = stream.next().await {
         match event {
             ResponseEvent::Created => {
                 let mut state = sess.state.lock().unwrap();
@@ -1172,6 +1218,20 @@ async fn try_run_turn(
                 state.previous_response_id = Some(response_id);
                 break;
             }
+            ResponseEvent::OutputTextDelta(delta) => {
+                let event = Event {
+                    id: sub_id.to_string(),
+                    msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta }),
+                };
+                sess.tx_event.send(event).await.ok();
+            }
+            ResponseEvent::ReasoningSummaryDelta(delta) => {
+                let event = Event {
+                    id: sub_id.to_string(),
+                    msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta }),
+                };
+                sess.tx_event.send(event).await.ok();
+            }
         }
     }
     Ok(output)

codex-rs/core/src/mcp_connection_manager.rs

@@ -79,9 +79,19 @@ impl McpConnectionManager {
 
         // Launch all configured servers concurrently.
         let mut join_set = JoinSet::new();
+        let mut errors = ClientStartErrors::new();
 
         for (server_name, cfg) in mcp_servers {
-            // TODO: Verify server name: require `^[a-zA-Z0-9_-]+$`?
+            // Validate server name before spawning
+            if !is_valid_mcp_server_name(&server_name) {
+                let error = anyhow::anyhow!(
+                    "invalid server name '{}': must match pattern ^[a-zA-Z0-9_-]+$",
+                    server_name
+                );
+                errors.insert(server_name, error);
+                continue;
+            }
+
             join_set.spawn(async move {
                 let McpServerConfig { command, args, env } = cfg;
                 let client_res = McpClient::new_stdio_client(command, args, env).await;
@@ -117,7 +127,6 @@ impl McpConnectionManager {
 
         let mut clients: HashMap<String, std::sync::Arc<McpClient>> =
             HashMap::with_capacity(join_set.len());
-        let mut errors = ClientStartErrors::new();
 
         while let Some(res) = join_set.join_next().await {
             let (server_name, client_res) = res?; // JoinError propagation
@@ -208,3 +217,10 @@ pub async fn list_all_tools(
 
     Ok(aggregated)
 }
+
+fn is_valid_mcp_server_name(server_name: &str) -> bool {
+    !server_name.is_empty()
+        && server_name
+            .chars()
+            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
+}

codex-rs/core/src/protocol.rs

@@ -97,6 +97,9 @@ pub enum Op {
         decision: ReviewDecision,
     },
 
+    /// Load a previously saved session from disk and resume from it.
+    LoadSession { path: std::path::PathBuf },
+
     /// Append an entry to the persistent cross-session message history.
     ///
     /// Note the entry is not guaranteed to be logged if the user has
@@ -282,9 +285,15 @@ pub enum EventMsg {
     /// Agent text output message
     AgentMessage(AgentMessageEvent),
 
+    /// Agent text output delta message
+    AgentMessageDelta(AgentMessageDeltaEvent),
+
     /// Reasoning event from agent.
     AgentReasoning(AgentReasoningEvent),
 
+    /// Agent reasoning delta event from agent.
+    AgentReasoningDelta(AgentReasoningDeltaEvent),
+
     /// Ack the client's configure message.
     SessionConfigured(SessionConfiguredEvent),
 
@@ -340,11 +349,21 @@ pub struct AgentMessageEvent {
     pub message: String,
 }
 
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct AgentMessageDeltaEvent {
+    pub delta: String,
+}
+
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct AgentReasoningEvent {
     pub text: String,
 }
 
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct AgentReasoningDeltaEvent {
+    pub delta: String,
+}
+
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct McpToolCallBeginEvent {
     /// Identifier so this can be paired with the McpToolCallEnd event.

codex-rs/core/src/rollout.rs

@@ -7,11 +7,11 @@ use std::fs::File;
 use std::fs::{self};
 use std::io::Error as IoError;
 
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 use time::OffsetDateTime;
 use time::format_description::FormatItem;
 use time::macros::format_description;
-use tokio::io::AsyncWriteExt;
+use tokio::io::{AsyncSeekExt, AsyncWriteExt};
 use tokio::sync::mpsc::Sender;
 use tokio::sync::mpsc::{self};
 use uuid::Uuid;
@@ -22,26 +22,60 @@ use crate::models::ResponseItem;
 /// Folder inside `~/.codex` that holds saved rollouts.
 const SESSIONS_SUBDIR: &str = "sessions";
 
-#[derive(Serialize)]
-struct SessionMeta {
-    id: String,
-    timestamp: String,
+#[derive(Serialize, Deserialize, Clone, Default)]
+pub struct SessionMeta {
+    pub id: String,
+    pub timestamp: String,
     #[serde(skip_serializing_if = "Option::is_none")]
-    instructions: Option<String>,
+    pub instructions: Option<String>,
+}
+
+#[derive(Serialize, Deserialize, Default, Clone)]
+pub struct SessionStateSnapshot {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub previous_response_id: Option<String>,
+}
+
+#[derive(Serialize, Deserialize, Default, Clone)]
+pub struct SavedSession {
+    pub session: SessionMeta,
+    #[serde(default)]
+    pub items: Vec<ResponseItem>,
+    #[serde(default)]
+    pub state: SessionStateSnapshot,
 }
 
 /// Records all [`ResponseItem`]s for a session and flushes them to disk after
 /// every update.
 ///
-/// Rollouts are recorded as JSONL and can be inspected with tools such as:
+/// Rollouts are recorded as JSON and can be inspected with tools such as:
 ///
 /// ```ignore
-/// $ jq -C . ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl
-/// $ fx ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl
+/// $ jq -C . ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.json
+/// $ fx ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.json
 /// ```
 #[derive(Clone)]
 pub(crate) struct RolloutRecorder {
-    tx: Sender<String>,
+    tx: Sender<RolloutCmd>,
+}
+
+#[derive(Clone)]
+enum RolloutCmd {
+    AddItems(Vec<ResponseItem>),
+    UpdateState(SessionStateSnapshot),
+}
+
+async fn write_session(file: &mut tokio::fs::File, data: &SavedSession) {
+    if file.seek(std::io::SeekFrom::Start(0)).await.is_err() {
+        return;
+    }
+    if file.set_len(0).await.is_err() {
+        return;
+    }
+    if let Ok(json) = serde_json::to_vec_pretty(data) {
+        let _ = file.write_all(&json).await;
+    }
+    let _ = file.flush().await;
 }
 
 impl RolloutRecorder {
@@ -76,68 +110,92 @@ impl RolloutRecorder {
         // A reasonably-sized bounded channel. If the buffer fills up the send
         // future will yield, which is fine – we only need to ensure we do not
         // perform *blocking* I/O on the caller’s thread.
-        let (tx, mut rx) = mpsc::channel::<String>(256);
+        let (tx, mut rx) = mpsc::channel::<RolloutCmd>(256);
+
+        let mut data = SavedSession {
+            session: meta,
+            items: Vec::new(),
+            state: SessionStateSnapshot::default(),
+        };
 
-        // Spawn a Tokio task that owns the file handle and performs async
-        // writes. Using `tokio::fs::File` keeps everything on the async I/O
-        // driver instead of blocking the runtime.
         tokio::task::spawn(async move {
             let mut file = tokio::fs::File::from_std(file);
 
-            while let Some(line) = rx.recv().await {
-                // Write line + newline, then flush to disk.
-                if let Err(e) = file.write_all(line.as_bytes()).await {
-                    tracing::warn!("rollout writer: failed to write line: {e}");
-                    break;
-                }
-                if let Err(e) = file.write_all(b"\n").await {
-                    tracing::warn!("rollout writer: failed to write newline: {e}");
-                    break;
-                }
-                if let Err(e) = file.flush().await {
-                    tracing::warn!("rollout writer: failed to flush: {e}");
-                    break;
+            write_session(&mut file, &data).await;
+
+            while let Some(cmd) = rx.recv().await {
+                match cmd {
+                    RolloutCmd::AddItems(items) => data.items.extend(items),
+                    RolloutCmd::UpdateState(state) => data.state = state,
                 }
+                write_session(&mut file, &data).await;
             }
         });
 
         let recorder = Self { tx };
-        // Ensure SessionMeta is the first item in the file.
-        recorder.record_item(&meta).await?;
         Ok(recorder)
     }
 
     /// Append `items` to the rollout file.
     pub(crate) async fn record_items(&self, items: &[ResponseItem]) -> std::io::Result<()> {
+        let mut filtered = Vec::new();
         for item in items {
             match item {
-                // Note that function calls may look a bit strange if they are
-                // "fully qualified MCP tool calls," so we could consider
-                // reformatting them in that case.
                 ResponseItem::Message { .. }
                 | ResponseItem::LocalShellCall { .. }
                 | ResponseItem::FunctionCall { .. }
-                | ResponseItem::FunctionCallOutput { .. } => {}
-                ResponseItem::Reasoning { .. } | ResponseItem::Other => {
-                    // These should never be serialized.
-                    continue;
-                }
+                | ResponseItem::FunctionCallOutput { .. } => filtered.push(item.clone()),
+                ResponseItem::Reasoning { .. } | ResponseItem::Other => {}
             }
-            self.record_item(item).await?;
         }
-        Ok(())
+        if filtered.is_empty() {
+            return Ok(());
+        }
+        self.tx
+            .send(RolloutCmd::AddItems(filtered))
+            .await
+            .map_err(|e| IoError::other(format!("failed to queue rollout items: {e}")))
     }
 
-    async fn record_item(&self, item: &impl Serialize) -> std::io::Result<()> {
-        // Serialize the item to JSON first so that the writer thread only has
-        // to perform the actual write.
-        let json = serde_json::to_string(item)
-            .map_err(|e| IoError::other(format!("failed to serialize response items: {e}")))?;
-
+    pub(crate) async fn record_state(&self, state: SessionStateSnapshot) -> std::io::Result<()> {
         self.tx
-            .send(json)
+            .send(RolloutCmd::UpdateState(state))
             .await
-            .map_err(|e| IoError::other(format!("failed to queue rollout item: {e}")))
+            .map_err(|e| IoError::other(format!("failed to queue rollout state: {e}")))
+    }
+
+    pub async fn resume(path: &std::path::Path) -> std::io::Result<(Self, SavedSession)> {
+        let bytes = tokio::fs::read(path).await?;
+        let saved: SavedSession = serde_json::from_slice(&bytes)
+            .map_err(|e| IoError::other(format!("failed to parse session: {e}")))?;
+
+        let file = std::fs::OpenOptions::new()
+            .write(true)
+            .read(true)
+            .open(path)?;
+        let saved_clone = saved.clone();
+        let (tx, mut rx) = mpsc::channel::<RolloutCmd>(256);
+        tokio::task::spawn(async move {
+            let mut data = saved_clone;
+            let mut file = tokio::fs::File::from_std(file);
+            write_session(&mut file, &data).await;
+            while let Some(cmd) = rx.recv().await {
+                match cmd {
+                    RolloutCmd::AddItems(items) => data.items.extend(items),
+                    RolloutCmd::UpdateState(state) => data.state = state,
+                }
+                write_session(&mut file, &data).await;
+            }
+        });
+
+        Ok((Self { tx }, saved))
+    }
+
+    pub async fn load(path: &std::path::Path) -> std::io::Result<SavedSession> {
+        let bytes = tokio::fs::read(path).await?;
+        let saved: SavedSession = serde_json::from_slice(&bytes)
+            .map_err(|e| IoError::other(format!("failed to parse session: {e}")))?;
+        Ok(saved)
     }
 }
 
@@ -153,13 +211,15 @@ struct LogFileInfo {
 }
 
 fn create_log_file(config: &Config, session_id: Uuid) -> std::io::Result<LogFileInfo> {
-    // Resolve ~/.codex/sessions and create it if missing.
-    let mut dir = config.codex_home.clone();
-    dir.push(SESSIONS_SUBDIR);
-    fs::create_dir_all(&dir)?;
-
+    // Resolve ~/.codex/sessions/YYYY/MM/DD and create it if missing.
     let timestamp = OffsetDateTime::now_local()
         .map_err(|e| IoError::other(format!("failed to get local time: {e}")))?;
+    let mut dir = config.codex_home.clone();
+    dir.push(SESSIONS_SUBDIR);
+    dir.push(timestamp.year().to_string());
+    dir.push(format!("{:02}", u8::from(timestamp.month())));
+    dir.push(format!("{:02}", timestamp.day()));
+    fs::create_dir_all(&dir)?;
 
     // Custom format for YYYY-MM-DDThh-mm-ss. Use `-` instead of `:` for
     // compatibility with filesystems that do not allow colons in filenames.
@@ -169,12 +229,13 @@ fn create_log_file(config: &Config, session_id: Uuid) -> std::io::Result<LogFile
         .format(format)
         .map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;
 
-    let filename = format!("rollout-{date_str}-{session_id}.jsonl");
+    let filename = format!("rollout-{date_str}-{session_id}.json");
 
     let path = dir.join(filename);
     let file = std::fs::OpenOptions::new()
-        .append(true)
+        .write(true)
         .create(true)
+        .truncate(true)
         .open(&path)?;
 
     Ok(LogFileInfo {

codex-rs/core/tests/cli_stream.rs

@@ -2,7 +2,12 @@
 
 use assert_cmd::Command as AssertCommand;
 use codex_core::exec::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR;
+use serde_json::Value;
+use std::time::Duration;
+use std::time::Instant;
 use tempfile::TempDir;
+use uuid::Uuid;
+use walkdir::WalkDir;
 use wiremock::Mock;
 use wiremock::MockServer;
 use wiremock::ResponseTemplate;
@@ -71,8 +76,8 @@ async fn chat_mode_stream_cli() {
     println!("Stderr:\n{}", String::from_utf8_lossy(&output.stderr));
     assert!(output.status.success());
     let stdout = String::from_utf8_lossy(&output.stdout);
-    assert!(stdout.contains("hi"));
-    assert_eq!(stdout.matches("hi").count(), 1);
+    let hi_lines = stdout.lines().filter(|line| line.trim() == "hi").count();
+    assert_eq!(hi_lines, 1, "Expected exactly one line with 'hi'");
 
     server.verify().await;
 }
@@ -117,3 +122,154 @@ async fn responses_api_stream_cli() {
     let stdout = String::from_utf8_lossy(&output.stdout);
     assert!(stdout.contains("fixture hello"));
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn integration_creates_and_checks_session_file() {
+    // Honor sandbox network restrictions for CI parity with the other tests.
+    if std::env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() {
+        println!(
+            "Skipping test because it cannot execute when network is disabled in a Codex sandbox."
+        );
+        return;
+    }
+
+    // 1. Temp home so we read/write isolated session files.
+    let home = TempDir::new().unwrap();
+
+    // 2. Unique marker we'll look for in the session log.
+    let marker = format!("integration-test-{}", Uuid::new_v4());
+    let prompt = format!("echo {marker}");
+
+    // 3. Use the same offline SSE fixture as responses_api_stream_cli so the test is hermetic.
+    let fixture =
+        std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/cli_responses_fixture.sse");
+
+    // 4. Run the codex CLI through cargo (ensures the right bin is built) and invoke `exec`,
+    //    which is what records a session.
+    let mut cmd = AssertCommand::new("cargo");
+    cmd.arg("run")
+        .arg("-p")
+        .arg("codex-cli")
+        .arg("--quiet")
+        .arg("--")
+        .arg("exec")
+        .arg("--skip-git-repo-check")
+        .arg("-C")
+        .arg(env!("CARGO_MANIFEST_DIR"))
+        .arg(&prompt);
+    cmd.env("CODEX_HOME", home.path())
+        .env("OPENAI_API_KEY", "dummy")
+        .env("CODEX_RS_SSE_FIXTURE", &fixture)
+        // Required for CLI arg parsing even though fixture short-circuits network usage.
+        .env("OPENAI_BASE_URL", "http://unused.local");
+
+    let output = cmd.output().unwrap();
+    assert!(
+        output.status.success(),
+        "codex-cli exec failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+
+    // 5. Sessions are written asynchronously; wait briefly for the directory to appear.
+    let sessions_dir = home.path().join("sessions");
+    let start = Instant::now();
+    while !sessions_dir.exists() && start.elapsed() < Duration::from_secs(2) {
+        std::thread::sleep(Duration::from_millis(50));
+    }
+
+    // 6. Scan all session files and find the one that contains our marker.
+    let mut matching_files = vec![];
+    for entry in WalkDir::new(&sessions_dir) {
+        let entry = entry.unwrap();
+        if entry.file_type().is_file() && entry.file_name().to_string_lossy().ends_with(".jsonl") {
+            let path = entry.path();
+            let content = std::fs::read_to_string(path).unwrap();
+            let mut lines = content.lines();
+            // Skip SessionMeta (first line)
+            let _ = lines.next();
+            for line in lines {
+                let item: Value = serde_json::from_str(line).unwrap();
+                if let Some("message") = item.get("type").and_then(|t| t.as_str()) {
+                    if let Some(content) = item.get("content") {
+                        if content.to_string().contains(&marker) {
+                            matching_files.push(path.to_owned());
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+    }
+    assert_eq!(
+        matching_files.len(),
+        1,
+        "Expected exactly one session file containing the marker, found {}",
+        matching_files.len()
+    );
+    let path = &matching_files[0];
+
+    // 7. Verify directory structure: sessions/YYYY/MM/DD/filename.jsonl
+    let rel = match path.strip_prefix(&sessions_dir) {
+        Ok(r) => r,
+        Err(_) => panic!("session file should live under sessions/"),
+    };
+    let comps: Vec<String> = rel
+        .components()
+        .map(|c| c.as_os_str().to_string_lossy().into_owned())
+        .collect();
+    assert_eq!(
+        comps.len(),
+        4,
+        "Expected sessions/YYYY/MM/DD/<file>, got {rel:?}"
+    );
+    let year = &comps[0];
+    let month = &comps[1];
+    let day = &comps[2];
+    assert!(
+        year.len() == 4 && year.chars().all(|c| c.is_ascii_digit()),
+        "Year dir not 4-digit numeric: {year}"
+    );
+    assert!(
+        month.len() == 2 && month.chars().all(|c| c.is_ascii_digit()),
+        "Month dir not zero-padded 2-digit numeric: {month}"
+    );
+    assert!(
+        day.len() == 2 && day.chars().all(|c| c.is_ascii_digit()),
+        "Day dir not zero-padded 2-digit numeric: {day}"
+    );
+    // Range checks (best-effort; won't fail on leading zeros)
+    if let Ok(m) = month.parse::<u8>() {
+        assert!((1..=12).contains(&m), "Month out of range: {m}");
+    }
+    if let Ok(d) = day.parse::<u8>() {
+        assert!((1..=31).contains(&d), "Day out of range: {d}");
+    }
+
+    // 8. Parse SessionMeta line and basic sanity checks.
+    let content = std::fs::read_to_string(path).unwrap();
+    let mut lines = content.lines();
+    let meta: Value = serde_json::from_str(lines.next().unwrap()).unwrap();
+    assert!(meta.get("id").is_some(), "SessionMeta missing id");
+    assert!(
+        meta.get("timestamp").is_some(),
+        "SessionMeta missing timestamp"
+    );
+
+    // 9. Confirm at least one message contains the marker.
+    let mut found_message = false;
+    for line in lines {
+        let item: Value = serde_json::from_str(line).unwrap();
+        if item.get("type").map(|t| t == "message").unwrap_or(false) {
+            if let Some(content) = item.get("content") {
+                if content.to_string().contains(&marker) {
+                    found_message = true;
+                    break;
+                }
+            }
+        }
+    }
+    assert!(
+        found_message,
+        "No message found in session file containing the marker"
+    );
+}

codex-rs/core/tests/stream_no_completed.rs

@@ -32,6 +32,8 @@ fn sse_completed(id: &str) -> String {
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+// this test is flaky (has race conditions), so we ignore it for now
+#[ignore]
 async fn retries_on_early_close() {
     #![allow(clippy::unwrap_used)]
 

codex-rs/exec/src/event_processor.rs

@@ -3,7 +3,9 @@ use codex_common::summarize_sandbox_policy;
 use codex_core::WireApi;
 use codex_core::config::Config;
 use codex_core::model_supports_reasoning_summaries;
+use codex_core::protocol::AgentMessageDeltaEvent;
 use codex_core::protocol::AgentMessageEvent;
+use codex_core::protocol::AgentReasoningDeltaEvent;
 use codex_core::protocol::BackgroundEventEvent;
 use codex_core::protocol::ErrorEvent;
 use codex_core::protocol::Event;
@@ -21,6 +23,7 @@ use owo_colors::OwoColorize;
 use owo_colors::Style;
 use shlex::try_join;
 use std::collections::HashMap;
+use std::io::Write;
 use std::time::Instant;
 
 /// This should be configurable. When used in CI, users may not want to impose
@@ -50,10 +53,12 @@ pub(crate) struct EventProcessor {
 
     /// Whether to include `AgentReasoning` events in the output.
     show_agent_reasoning: bool,
+    answer_started: bool,
+    reasoning_started: bool,
 }
 
 impl EventProcessor {
-    pub(crate) fn create_with_ansi(with_ansi: bool, show_agent_reasoning: bool) -> Self {
+    pub(crate) fn create_with_ansi(with_ansi: bool, config: &Config) -> Self {
         let call_id_to_command = HashMap::new();
         let call_id_to_patch = HashMap::new();
         let call_id_to_tool_call = HashMap::new();
@@ -70,7 +75,9 @@ impl EventProcessor {
                 green: Style::new().green(),
                 cyan: Style::new().cyan(),
                 call_id_to_tool_call,
-                show_agent_reasoning,
+                show_agent_reasoning: !config.hide_agent_reasoning,
+                answer_started: false,
+                reasoning_started: false,
             }
         } else {
             Self {
@@ -84,7 +91,9 @@ impl EventProcessor {
                 green: Style::new(),
                 cyan: Style::new(),
                 call_id_to_tool_call,
-                show_agent_reasoning,
+                show_agent_reasoning: !config.hide_agent_reasoning,
+                answer_started: false,
+                reasoning_started: false,
             }
         }
     }
@@ -184,12 +193,45 @@ impl EventProcessor {
             EventMsg::TokenCount(TokenUsage { total_tokens, .. }) => {
                 ts_println!(self, "tokens used: {total_tokens}");
             }
+            EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta }) => {
+                if !self.answer_started {
+                    ts_println!(self, "{}\n", "codex".style(self.italic).style(self.magenta));
+                    self.answer_started = true;
+                }
+                print!("{delta}");
+                #[allow(clippy::expect_used)]
+                std::io::stdout().flush().expect("could not flush stdout");
+            }
+            EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta }) => {
+                if !self.show_agent_reasoning {
+                    return;
+                }
+                if !self.reasoning_started {
+                    ts_println!(
+                        self,
+                        "{}\n",
+                        "thinking".style(self.italic).style(self.magenta),
+                    );
+                    self.reasoning_started = true;
+                }
+                print!("{delta}");
+                #[allow(clippy::expect_used)]
+                std::io::stdout().flush().expect("could not flush stdout");
+            }
             EventMsg::AgentMessage(AgentMessageEvent { message }) => {
-                ts_println!(
-                    self,
-                    "{}\n{message}",
-                    "codex".style(self.bold).style(self.magenta)
-                );
+                // if answer_started is false, this means we haven't received any
+                // delta. Thus, we need to print the message as a new answer.
+                if !self.answer_started {
+                    ts_println!(
+                        self,
+                        "{}\n{}",
+                        "codex".style(self.italic).style(self.magenta),
+                        message,
+                    );
+                } else {
+                    println!();
+                    self.answer_started = false;
+                }
             }
             EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
                 call_id,
@@ -343,7 +385,7 @@ impl EventProcessor {
                 );
 
                 // Pretty-print the patch summary with colored diff markers so
-                // it’s easy to scan in the terminal output.
+                // it's easy to scan in the terminal output.
                 for (path, change) in changes.iter() {
                     match change {
                         FileChange::Add { content } => {
@@ -441,12 +483,17 @@ impl EventProcessor {
             }
             EventMsg::AgentReasoning(agent_reasoning_event) => {
                 if self.show_agent_reasoning {
-                    ts_println!(
-                        self,
-                        "{}\n{}",
-                        "thinking".style(self.italic).style(self.magenta),
-                        agent_reasoning_event.text
-                    );
+                    if !self.reasoning_started {
+                        ts_println!(
+                            self,
+                            "{}\n{}",
+                            "codex".style(self.italic).style(self.magenta),
+                            agent_reasoning_event.text,
+                        );
+                    } else {
+                        println!();
+                        self.reasoning_started = false;
+                    }
                 }
             }
             EventMsg::SessionConfigured(session_configured_event) => {

codex-rs/exec/src/lib.rs

@@ -115,8 +115,7 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
     };
 
     let config = Config::load_with_cli_overrides(cli_kv_overrides, overrides)?;
-    let mut event_processor =
-        EventProcessor::create_with_ansi(stdout_with_ansi, !config.hide_agent_reasoning);
+    let mut event_processor = EventProcessor::create_with_ansi(stdout_with_ansi, &config);
     // Print the effective configuration and prompt so users can see what Codex
     // is using.
     event_processor.print_config_summary(&config, &prompt);

codex-rs/mcp-server/src/codex_tool_runner.rs

@@ -171,6 +171,12 @@ pub async fn run_codex_tool_session(
                     EventMsg::SessionConfigured(_) => {
                         tracing::error!("unexpected SessionConfigured event");
                     }
+                    EventMsg::AgentMessageDelta(_) => {
+                        // TODO: think how we want to support this in the MCP
+                    }
+                    EventMsg::AgentReasoningDelta(_) => {
+                        // TODO: think how we want to support this in the MCP
+                    }
                     EventMsg::Error(_)
                     | EventMsg::TaskStarted
                     | EventMsg::TokenCount(_)

codex-rs/tui/src/app.rs

@@ -199,7 +199,21 @@ impl<'a> App<'a> {
                             modifiers: crossterm::event::KeyModifiers::CONTROL,
                             ..
                         } => {
-                            self.app_event_tx.send(AppEvent::ExitRequest);
+                            match &mut self.app_state {
+                                AppState::Chat { widget } => {
+                                    if widget.composer_is_empty() {
+                                        self.app_event_tx.send(AppEvent::ExitRequest);
+                                    } else {
+                                        // Treat Ctrl+D as a normal key event when the composer
+                                        // is not empty so that it doesn't quit the application
+                                        // prematurely.
+                                        self.dispatch_key_event(key_event);
+                                    }
+                                }
+                                AppState::Login { .. } | AppState::GitWarning { .. } => {
+                                    self.app_event_tx.send(AppEvent::ExitRequest);
+                                }
+                            }
                         }
                         _ => {
                             self.dispatch_key_event(key_event);
@@ -283,6 +297,8 @@ impl<'a> App<'a> {
     }
 
     fn draw_next_frame(&mut self, terminal: &mut tui::Tui) -> Result<()> {
+        // TODO: add a throttle to avoid redrawing too often
+
         match &mut self.app_state {
             AppState::Chat { widget } => {
                 terminal.draw(|frame| frame.render_widget_ref(&**widget, frame.area()))?;

codex-rs/tui/src/bottom_pane/chat_composer.rs

@@ -76,6 +76,11 @@ impl ChatComposer<'_> {
         this
     }
 
+    /// Returns true if the composer currently contains no user input.
+    pub(crate) fn is_empty(&self) -> bool {
+        self.textarea.is_empty()
+    }
+
     /// Update the cached *context-left* percentage and refresh the placeholder
     /// text. The UI relies on the placeholder to convey the remaining
     /// context when the composer is empty.

codex-rs/tui/src/bottom_pane/chat_composer_history.rs

@@ -72,8 +72,7 @@ impl ChatComposerHistory {
             return false;
         }
 
-        let lines = textarea.lines();
-        if lines.len() == 1 && lines[0].is_empty() {
+        if textarea.is_empty() {
             return true;
         }
 
@@ -85,6 +84,7 @@ impl ChatComposerHistory {
             return false;
         }
 
+        let lines = textarea.lines();
         matches!(&self.last_history_text, Some(prev) if prev == &lines.join("\n"))
     }
 

codex-rs/tui/src/bottom_pane/mod.rs

@@ -162,6 +162,10 @@ impl BottomPane<'_> {
         }
     }
 
+    pub(crate) fn composer_is_empty(&self) -> bool {
+        self.composer.is_empty()
+    }
+
     pub(crate) fn is_task_running(&self) -> bool {
         self.is_task_running
     }

codex-rs/tui/src/chatwidget.rs

@@ -3,7 +3,9 @@ use std::sync::Arc;
 
 use codex_core::codex_wrapper::init_codex;
 use codex_core::config::Config;
+use codex_core::protocol::AgentMessageDeltaEvent;
 use codex_core::protocol::AgentMessageEvent;
+use codex_core::protocol::AgentReasoningDeltaEvent;
 use codex_core::protocol::AgentReasoningEvent;
 use codex_core::protocol::ApplyPatchApprovalRequestEvent;
 use codex_core::protocol::ErrorEvent;
@@ -49,6 +51,8 @@ pub(crate) struct ChatWidget<'a> {
     config: Config,
     initial_user_message: Option<UserMessage>,
     token_usage: TokenUsage,
+    reasoning_buffer: String,
+    answer_buffer: String,
 }
 
 #[derive(Clone, Copy, Eq, PartialEq)]
@@ -135,6 +139,8 @@ impl ChatWidget<'_> {
                 initial_images,
             ),
             token_usage: TokenUsage::default(),
+            reasoning_buffer: String::new(),
+            answer_buffer: String::new(),
         }
     }
 
@@ -240,16 +246,51 @@ impl ChatWidget<'_> {
                 self.request_redraw();
             }
             EventMsg::AgentMessage(AgentMessageEvent { message }) => {
+                // if the answer buffer is empty, this means we haven't received any
+                // delta. Thus, we need to print the message as a new answer.
+                if self.answer_buffer.is_empty() {
+                    self.conversation_history
+                        .add_agent_message(&self.config, message);
+                } else {
+                    self.conversation_history
+                        .replace_prev_agent_message(&self.config, message);
+                }
+                self.answer_buffer.clear();
+                self.request_redraw();
+            }
+            EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta }) => {
+                if self.answer_buffer.is_empty() {
+                    self.conversation_history
+                        .add_agent_message(&self.config, "".to_string());
+                }
+                self.answer_buffer.push_str(&delta.clone());
                 self.conversation_history
-                    .add_agent_message(&self.config, message);
+                    .replace_prev_agent_message(&self.config, self.answer_buffer.clone());
+                self.request_redraw();
+            }
+            EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent { delta }) => {
+                if self.reasoning_buffer.is_empty() {
+                    self.conversation_history
+                        .add_agent_reasoning(&self.config, "".to_string());
+                }
+                self.reasoning_buffer.push_str(&delta.clone());
+                self.conversation_history
+                    .replace_prev_agent_reasoning(&self.config, self.reasoning_buffer.clone());
                 self.request_redraw();
             }
             EventMsg::AgentReasoning(AgentReasoningEvent { text }) => {
-                if !self.config.hide_agent_reasoning {
+                // if the reasoning buffer is empty, this means we haven't received any
+                // delta. Thus, we need to print the message as a new reasoning.
+                if self.reasoning_buffer.is_empty() {
                     self.conversation_history
-                        .add_agent_reasoning(&self.config, text);
-                    self.request_redraw();
+                        .add_agent_reasoning(&self.config, "".to_string());
+                } else {
+                    // else, we rerender one last time.
+                    self.conversation_history
+                        .replace_prev_agent_reasoning(&self.config, text);
                 }
+                self.reasoning_buffer.clear();
+                self.request_redraw();
             }
             EventMsg::TaskStarted => {
                 self.bottom_pane.clear_ctrl_c_quit_hint();
@@ -432,6 +473,10 @@ impl ChatWidget<'_> {
         }
     }
 
+    pub(crate) fn composer_is_empty(&self) -> bool {
+        self.bottom_pane.composer_is_empty()
+    }
+
     /// Forward an `Op` directly to codex.
     pub(crate) fn submit_op(&self, op: Op) {
         if let Err(e) = self.codex_op_tx.send(op) {

codex-rs/tui/src/conversation_history_widget.rs

@@ -202,6 +202,14 @@ impl ConversationHistoryWidget {
         self.add_to_history(HistoryCell::new_agent_reasoning(config, text));
     }
 
+    pub fn replace_prev_agent_reasoning(&mut self, config: &Config, text: String) {
+        self.replace_last_agent_reasoning(config, text);
+    }
+
+    pub fn replace_prev_agent_message(&mut self, config: &Config, text: String) {
+        self.replace_last_agent_message(config, text);
+    }
+
     pub fn add_background_event(&mut self, message: String) {
         self.add_to_history(HistoryCell::new_background_event(message));
     }
@@ -249,6 +257,42 @@ impl ConversationHistoryWidget {
         });
     }
 
+    pub fn replace_last_agent_reasoning(&mut self, config: &Config, text: String) {
+        if let Some(idx) = self
+            .entries
+            .iter()
+            .rposition(|entry| matches!(entry.cell, HistoryCell::AgentReasoning { .. }))
+        {
+            let width = self.cached_width.get();
+            let entry = &mut self.entries[idx];
+            entry.cell = HistoryCell::new_agent_reasoning(config, text);
+            let height = if width > 0 {
+                entry.cell.height(width)
+            } else {
+                0
+            };
+            entry.line_count.set(height);
+        }
+    }
+
+    pub fn replace_last_agent_message(&mut self, config: &Config, text: String) {
+        if let Some(idx) = self
+            .entries
+            .iter()
+            .rposition(|entry| matches!(entry.cell, HistoryCell::AgentMessage { .. }))
+        {
+            let width = self.cached_width.get();
+            let entry = &mut self.entries[idx];
+            entry.cell = HistoryCell::new_agent_message(config, text);
+            let height = if width > 0 {
+                entry.cell.height(width)
+            } else {
+                0
+            };
+            entry.line_count.set(height);
+        }
+    }
+
     pub fn record_completed_exec_command(
         &mut self,
         call_id: String,
@@ -454,7 +498,7 @@ impl WidgetRef for ConversationHistoryWidget {
 
         {
             // Choose a thumb color that stands out only when this pane has focus so that the
-            // user’s attention is naturally drawn to the active viewport. When unfocused we show
+            // user's attention is naturally drawn to the active viewport. When unfocused we show
             // a low-contrast thumb so the scrollbar fades into the background without becoming
             // invisible.
             let thumb_style = if self.has_input_focus {