diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 6129c0de2..af746b37b 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -2,7 +2,9 @@ use anyhow::Result; use app_test_support::McpProcess; use app_test_support::create_fake_rollout_with_text_elements; use app_test_support::create_mock_responses_server_repeating_assistant; +use app_test_support::rollout_path; use app_test_support::to_response; +use chrono::Utc; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::SessionSource; @@ -22,6 +24,8 @@ use codex_protocol::user_input::TextElement; use core_test_support::responses; use core_test_support::skip_if_no_network; use pretty_assertions::assert_eq; +use std::fs::FileTimes; +use std::path::Path; use std::path::PathBuf; use tempfile::TempDir; use tokio::time::timeout; @@ -147,6 +151,116 @@ async fn thread_resume_returns_rollout_history() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_resume_without_overrides_does_not_change_updated_at_or_mtime() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + let rollout = setup_rollout_fixture(codex_home.path(), &server.uri())?; + let thread_id = rollout.conversation_id.clone(); + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread_id.clone(), + ..Default::default() + }) + .await?; + let resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), + ) + .await??; + let ThreadResumeResponse { thread, .. } = to_response::(resume_resp)?; + + assert_eq!(thread.updated_at, rollout.expected_updated_at); + + let after_modified = std::fs::metadata(&rollout.rollout_file_path)?.modified()?; + assert_eq!(after_modified, rollout.before_modified); + + let turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id, + input: vec![UserInput::Text { + text: "Hello".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_id)), + ) + .await??; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let after_turn_modified = std::fs::metadata(&rollout.rollout_file_path)?.modified()?; + assert!(after_turn_modified > rollout.before_modified); + + Ok(()) +} + +#[tokio::test] +async fn thread_resume_with_overrides_defers_updated_at_until_turn_start() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + let rollout = setup_rollout_fixture(codex_home.path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: rollout.conversation_id.clone(), + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), + ) + .await??; + let ThreadResumeResponse { thread, .. } = to_response::(resume_resp)?; + + assert_eq!(thread.updated_at, rollout.expected_updated_at); + + let after_resume_modified = std::fs::metadata(&rollout.rollout_file_path)?.modified()?; + assert_eq!(after_resume_modified, rollout.before_modified); + + let turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: rollout.conversation_id, + input: vec![UserInput::Text { + text: "Hello".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_id)), + ) + .await??; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let after_turn_modified = std::fs::metadata(&rollout.rollout_file_path)?.modified()?; + assert!(after_turn_modified > rollout.before_modified); + + Ok(()) +} + #[tokio::test] async fn thread_resume_prefers_path_over_thread_id() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; @@ -364,3 +478,51 @@ stream_max_retries = 0 ), ) } + +fn set_rollout_mtime(path: &Path, updated_at_rfc3339: &str) -> Result<()> { + let parsed = chrono::DateTime::parse_from_rfc3339(updated_at_rfc3339)?.with_timezone(&Utc); + let times = FileTimes::new().set_modified(parsed.into()); + std::fs::OpenOptions::new() + .append(true) + .open(path)? + .set_times(times)?; + Ok(()) +} + +struct RolloutFixture { + conversation_id: String, + rollout_file_path: PathBuf, + before_modified: std::time::SystemTime, + expected_updated_at: i64, +} + +fn setup_rollout_fixture(codex_home: &Path, server_uri: &str) -> Result { + create_config_toml(codex_home, server_uri)?; + + let preview = "Saved user message"; + let filename_ts = "2025-01-05T12-00-00"; + let meta_rfc3339 = "2025-01-05T12:00:00Z"; + let expected_updated_at_rfc3339 = "2025-01-07T00:00:00Z"; + let conversation_id = create_fake_rollout_with_text_elements( + codex_home, + filename_ts, + meta_rfc3339, + preview, + Vec::new(), + Some("mock_provider"), + None, + )?; + let rollout_file_path = rollout_path(codex_home, filename_ts, &conversation_id); + set_rollout_mtime(rollout_file_path.as_path(), expected_updated_at_rfc3339)?; + let before_modified = std::fs::metadata(&rollout_file_path)?.modified()?; + let expected_updated_at = chrono::DateTime::parse_from_rfc3339(expected_updated_at_rfc3339)? + .with_timezone(&Utc) + .timestamp(); + + Ok(RolloutFixture { + conversation_id, + rollout_file_path, + before_modified, + expected_updated_at, + }) +} diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 41e148e11..26cfdefea 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -933,23 +933,28 @@ impl Session { // Build and record initial items (user instructions + environment context) let items = self.build_initial_context(&turn_context).await; self.record_conversation_items(&turn_context, &items).await; + { + let mut state = self.state.lock().await; + state.initial_context_seeded = true; + } // Ensure initial items are visible to immediate readers (e.g., tests, forks). self.flush_rollout().await; } - InitialHistory::Resumed(_) | InitialHistory::Forked(_) => { - let rollout_items = conversation_history.get_rollout_items(); - let persist = matches!(conversation_history, InitialHistory::Forked(_)); + InitialHistory::Resumed(resumed_history) => { + let rollout_items = resumed_history.history; + { + let mut state = self.state.lock().await; + state.initial_context_seeded = false; + } // If resuming, warn when the last recorded model differs from the current one. - if let InitialHistory::Resumed(_) = conversation_history - && let Some(prev) = rollout_items.iter().rev().find_map(|it| { - if let RolloutItem::TurnContext(ctx) = it { - Some(ctx.model.as_str()) - } else { - None - } - }) - { + if let Some(prev) = rollout_items.iter().rev().find_map(|it| { + if let RolloutItem::TurnContext(ctx) = it { + Some(ctx.model.as_str()) + } else { + None + } + }) { let curr = turn_context.client.get_model(); if prev != curr { warn!( @@ -984,8 +989,29 @@ impl Session { state.set_token_info(Some(info)); } + // Defer seeding the session's initial context until the first turn starts so + // turn/start overrides can be merged before we write to the rollout. + self.flush_rollout().await; + } + InitialHistory::Forked(rollout_items) => { + // Always add response items to conversation history + let reconstructed_history = self + .reconstruct_history_from_rollout(&turn_context, &rollout_items) + .await; + if !reconstructed_history.is_empty() { + self.record_into_history(&reconstructed_history, &turn_context) + .await; + } + + // Seed usage info from the recorded rollout so UIs can show token counts + // immediately on resume/fork. + if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) { + let mut state = self.state.lock().await; + state.set_token_info(Some(info)); + } + // If persisting, persist all rollout items as-is (recorder filters) - if persist && !rollout_items.is_empty() { + if !rollout_items.is_empty() { self.persist_rollout_items(&rollout_items).await; } @@ -993,6 +1019,10 @@ impl Session { let initial_context = self.build_initial_context(&turn_context).await; self.record_conversation_items(&turn_context, &initial_context) .await; + { + let mut state = self.state.lock().await; + state.initial_context_seeded = true; + } // Flush after seeding history and any persisted rollout copy. self.flush_rollout().await; } @@ -1643,6 +1673,21 @@ impl Session { state.replace_history(items); } + pub(crate) async fn seed_initial_context_if_needed(&self, turn_context: &TurnContext) { + { + let mut state = self.state.lock().await; + if state.initial_context_seeded { + return; + } + state.initial_context_seeded = true; + } + + let initial_context = self.build_initial_context(turn_context).await; + self.record_conversation_items(turn_context, &initial_context) + .await; + self.flush_rollout().await; + } + async fn persist_rollout_response_items(&self, items: &[ResponseItem]) { let rollout_items: Vec = items .iter() @@ -2332,6 +2377,11 @@ mod handlers { return; } + let initial_context_seeded = sess.state.lock().await.initial_context_seeded; + if !initial_context_seeded { + return; + } + let current_context = sess.new_default_turn_with_sub_id(sub_id).await; let update_items = sess.build_settings_update_items( Some(&previous_context), @@ -2419,6 +2469,7 @@ mod handlers { // Attempt to inject input into current task if let Err(items) = sess.inject_input(items).await { + sess.seed_initial_context_if_needed(¤t_context).await; let update_items = sess.build_settings_update_items( previous_context.as_ref(), ¤t_context, @@ -3719,6 +3770,23 @@ mod tests { #[tokio::test] async fn record_initial_history_reconstructs_resumed_transcript() { + let (session, turn_context) = make_session_and_context().await; + let (rollout_items, expected) = sample_rollout(&session, &turn_context).await; + + session + .record_initial_history(InitialHistory::Resumed(ResumedHistory { + conversation_id: ThreadId::default(), + history: rollout_items, + rollout_path: PathBuf::from("/tmp/resume.jsonl"), + })) + .await; + + let history = session.state.lock().await.clone_history(); + assert_eq!(expected, history.raw_items()); + } + + #[tokio::test] + async fn resumed_history_seeds_initial_context_on_first_turn_only() { let (session, turn_context) = make_session_and_context().await; let (rollout_items, mut expected) = sample_rollout(&session, &turn_context).await; @@ -3730,9 +3798,17 @@ mod tests { })) .await; + let history_before_seed = session.state.lock().await.clone_history(); + assert_eq!(expected, history_before_seed.raw_items()); + + session.seed_initial_context_if_needed(&turn_context).await; expected.extend(session.build_initial_context(&turn_context).await); - let history = session.state.lock().await.clone_history(); - assert_eq!(expected, history.raw_items()); + let history_after_seed = session.clone_history().await; + assert_eq!(expected, history_after_seed.raw_items()); + + session.seed_initial_context_if_needed(&turn_context).await; + let history_after_second_seed = session.clone_history().await; + assert_eq!(expected, history_after_second_seed.raw_items()); } #[tokio::test] @@ -4347,7 +4423,8 @@ mod tests { session_configuration.session_source.clone(), ); - let state = SessionState::new(session_configuration.clone()); + let mut state = SessionState::new(session_configuration.clone()); + mark_state_initial_context_seeded(&mut state); let skills_manager = Arc::new(SkillsManager::new(config.codex_home.clone())); let services = SessionServices { @@ -4456,7 +4533,8 @@ mod tests { session_configuration.session_source.clone(), ); - let state = SessionState::new(session_configuration.clone()); + let mut state = SessionState::new(session_configuration.clone()); + mark_state_initial_context_seeded(&mut state); let skills_manager = Arc::new(SkillsManager::new(config.codex_home.clone())); let services = SessionServices { @@ -4502,6 +4580,10 @@ mod tests { (session, turn_context, rx_event) } + fn mark_state_initial_context_seeded(state: &mut SessionState) { + state.initial_context_seeded = true; + } + #[tokio::test] async fn refresh_mcp_servers_is_deferred_until_next_turn() { let (session, turn_context) = make_session_and_context().await; diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index 746396949..067f7b337 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -15,6 +15,11 @@ pub(crate) struct SessionState { pub(crate) history: ContextManager, pub(crate) latest_rate_limits: Option, pub(crate) server_reasoning_included: bool, + /// Whether the session's initial context has been seeded into history. + /// + /// TODO(owen): This is a temporary solution to avoid updating a thread's updated_at + /// timestamp when resuming a session. Remove this once SQLite is in place. + pub(crate) initial_context_seeded: bool, } impl SessionState { @@ -26,6 +31,7 @@ impl SessionState { history, latest_rate_limits: None, server_reasoning_included: false, + initial_context_seeded: false, } } diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index bbf9d9c27..08d23f798 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -115,6 +115,8 @@ impl Session { task: T, ) { self.abort_all_tasks(TurnAbortReason::Replaced).await; + self.seed_initial_context_if_needed(turn_context.as_ref()) + .await; let task: Arc = Arc::new(task); let task_kind = task.kind();