diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 0e622d7f5..981b52a59 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -332,6 +332,7 @@ use codex_protocol::protocol::McpServerRefreshConfig; use codex_protocol::protocol::Op; use codex_protocol::protocol::RateLimitSnapshot as CoreRateLimitSnapshot; use codex_protocol::protocol::RealtimeVoicesList; +use codex_protocol::protocol::ResumedHistory; use codex_protocol::protocol::ReviewDelivery as CoreReviewDelivery; use codex_protocol::protocol::ReviewRequest; use codex_protocol::protocol::ReviewTarget as CoreReviewTarget; @@ -354,6 +355,7 @@ use codex_state::log_db::LogDbLayer; use codex_thread_store::ArchiveThreadParams as StoreArchiveThreadParams; use codex_thread_store::ListThreadsParams as StoreListThreadsParams; use codex_thread_store::LocalThreadStore; +use codex_thread_store::ReadThreadByRolloutPathParams as StoreReadThreadByRolloutPathParams; use codex_thread_store::ReadThreadParams as StoreReadThreadParams; use codex_thread_store::RemoteThreadStore; use codex_thread_store::SortDirection as StoreSortDirection; @@ -404,7 +406,6 @@ use crate::thread_state::ThreadState; use crate::thread_state::ThreadStateManager; use token_usage_replay::latest_token_usage_turn_id_for_thread_path; use token_usage_replay::latest_token_usage_turn_id_from_rollout_items; -use token_usage_replay::latest_token_usage_turn_id_from_rollout_path; use token_usage_replay::send_thread_token_usage_update_to_connection; const THREAD_LIST_DEFAULT_LIMIT: usize = 25; @@ -662,9 +663,7 @@ pub(crate) struct CodexMessageProcessorArgs { fn configured_thread_store(config: &Config) -> Arc { match config.experimental_thread_store_endpoint.as_deref() { Some(endpoint) => Arc::new(RemoteThreadStore::new(endpoint)), - None => Arc::new(LocalThreadStore::new( - codex_rollout::RolloutConfig::from_view(config), - )), + None => Arc::new(configured_local_thread_store(config)), } } @@ -675,6 +674,10 @@ fn environment_selection_error_message(err: CodexErr) -> String { } } +fn configured_local_thread_store(config: &Config) -> LocalThreadStore { + LocalThreadStore::new(codex_rollout::RolloutConfig::from_view(config)) +} + impl CodexMessageProcessor { async fn instruction_sources_from_config(config: &Config) -> Vec { codex_core::AgentsMdManager::new(config) @@ -4509,22 +4512,22 @@ impl CodexMessageProcessor { } = params; let include_turns = !exclude_turns; - let thread_history = if let Some(history) = history { + let (thread_history, resume_source_thread) = if let Some(history) = history { let Some(thread_history) = self .resume_thread_from_history(request_id.clone(), history.as_slice()) .await else { return; }; - thread_history + (thread_history, None) } else { - let Some(thread_history) = self + let Some((thread_history, stored_thread)) = self .resume_thread_from_rollout(request_id.clone(), &thread_id, path.as_ref()) .await else { return; }; - thread_history + (thread_history, Some(stored_thread)) }; let history_cwd = thread_history.session_cwd(); @@ -4541,13 +4544,12 @@ impl CodexMessageProcessor { developer_instructions, personality, ); - let persisted_resume_metadata = self - .load_and_apply_persisted_resume_metadata( - &thread_history, - &mut request_overrides, - &mut typesafe_overrides, - ) - .await; + self.load_and_apply_persisted_resume_metadata( + &thread_history, + &mut request_overrides, + &mut typesafe_overrides, + ) + .await; // Derive a Config using the same logic as new conversation, honoring overrides if provided. let config = match self @@ -4612,7 +4614,7 @@ impl CodexMessageProcessor { codex_thread.as_ref(), &response_history, rollout_path.as_path(), - persisted_resume_metadata.as_ref(), + resume_source_thread, include_turns, ) .await @@ -4738,77 +4740,58 @@ impl CodexMessageProcessor { return true; } - let rollout_path = if let Some(path) = existing_thread.rollout_path() { - if path.exists() { - path - } else { - match find_thread_path_by_id_str( - &self.config.codex_home, - &existing_thread_id.to_string(), - ) - .await - { - Ok(Some(path)) => path, - Ok(None) => { - self.send_invalid_request_error( - request_id, - format!("no rollout found for thread id {existing_thread_id}"), - ) - .await; - return true; - } - Err(err) => { - self.send_invalid_request_error( - request_id, - format!("failed to locate thread id {existing_thread_id}: {err}"), - ) - .await; - return true; - } - } - } - } else { - match find_thread_path_by_id_str( - &self.config.codex_home, - &existing_thread_id.to_string(), - ) - .await - { - Ok(Some(path)) => path, - Ok(None) => { - self.send_invalid_request_error( - request_id, - format!("no rollout found for thread id {existing_thread_id}"), - ) - .await; - return true; - } - Err(err) => { - self.send_invalid_request_error( - request_id, - format!("failed to locate thread id {existing_thread_id}: {err}"), - ) - .await; - return true; - } - } - }; - - if let Some(requested_path) = params.path.as_ref() - && requested_path != &rollout_path + if let (Some(requested_path), Some(active_path)) = ( + params.path.as_ref(), + existing_thread.rollout_path().as_ref(), + ) && requested_path != active_path { self.send_invalid_request_error( request_id, format!( "cannot resume running thread {existing_thread_id} with mismatched path: requested `{}`, active `{}`", requested_path.display(), - rollout_path.display() + active_path.display() ), ) .await; return true; } + let Some(source_thread) = self + .read_stored_thread_for_resume( + request_id.clone(), + ¶ms.thread_id, + params.path.as_ref(), + /*include_history*/ true, + ) + .await + else { + return true; + }; + if source_thread.thread_id != existing_thread_id { + self.send_invalid_request_error( + request_id, + format!( + "cannot resume running thread {existing_thread_id} from source thread {}", + source_thread.thread_id + ), + ) + .await; + return true; + } + let Some(history_items) = source_thread + .history + .as_ref() + .map(|history| history.items.clone()) + else { + self.send_internal_error( + request_id, + format!("thread {existing_thread_id} did not include persisted history"), + ) + .await; + return true; + }; + let thread_state = self .thread_state_manager .thread_state(existing_thread_id) @@ -4835,18 +4818,15 @@ impl CodexMessageProcessor { mismatch_details.join("; ") ); } - let mut config_for_instruction_sources = self.config.as_ref().clone(); - config_for_instruction_sources.cwd = config_snapshot.cwd.clone(); - let instruction_sources = - Self::instruction_sources_from_config(&config_for_instruction_sources).await; - let thread_summary = match load_thread_summary_for_rollout( - &self.config, - existing_thread_id, - rollout_path.as_path(), - config_snapshot.model_provider_id.as_str(), - /*persisted_metadata*/ None, - ) - .await + let mut summary_source_thread = source_thread; + summary_source_thread.history = None; + let thread_summary = match self + .stored_thread_to_api_thread( + summary_source_thread, + config_snapshot.model_provider_id.as_str(), + /*include_turns*/ false, + ) + .await { Ok(thread) => thread, Err(message) => { @@ -4854,6 +4834,10 @@ impl CodexMessageProcessor { return true; } }; + let mut config_for_instruction_sources = self.config.as_ref().clone(); + config_for_instruction_sources.cwd = config_snapshot.cwd.clone(); + let instruction_sources = + Self::instruction_sources_from_config(&config_for_instruction_sources).await; let listener_command_tx = { let thread_state = thread_state.lock().await; @@ -4874,7 +4858,7 @@ impl CodexMessageProcessor { let command = crate::thread_state::ThreadListenerCommand::SendThreadResumeResponse( Box::new(crate::thread_state::PendingThreadResumeRequest { request_id: request_id.clone(), - rollout_path: rollout_path.clone(), + history_items, config_snapshot, instruction_sources, thread_summary, @@ -4920,57 +4904,133 @@ impl CodexMessageProcessor { request_id: ConnectionRequestId, thread_id: &str, path: Option<&PathBuf>, - ) -> Option { - let rollout_path = if let Some(path) = path { - path.clone() + ) -> Option<(InitialHistory, StoredThread)> { + match self + .read_stored_thread_for_resume( + request_id.clone(), + thread_id, + path, + /*include_history*/ true, + ) + .await + { + Some(stored_thread) => self + .stored_thread_to_initial_history(request_id, &stored_thread) + .await + .map(|history| (history, stored_thread)), + None => None, + } + } + + async fn read_stored_thread_for_resume( + &self, + request_id: ConnectionRequestId, + thread_id: &str, + path: Option<&PathBuf>, + include_history: bool, + ) -> Option { + let result = if let Some(path) = path { + self.thread_store + .read_thread_by_rollout_path(StoreReadThreadByRolloutPathParams { + rollout_path: path.clone(), + include_archived: true, + include_history, + }) + .await } else { let existing_thread_id = match ThreadId::from_string(thread_id) { Ok(id) => id, Err(err) => { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: format!("invalid thread id: {err}"), - data: None, - }; - self.outgoing.send_error(request_id, error).await; + self.send_invalid_request_error( + request_id, + format!("invalid thread id: {err}"), + ) + .await; return None; } }; - - match find_thread_path_by_id_str( - &self.config.codex_home, - &existing_thread_id.to_string(), - ) - .await - { - Ok(Some(path)) => path, - Ok(None) => { - self.send_invalid_request_error( - request_id, - format!("no rollout found for thread id {existing_thread_id}"), - ) - .await; - return None; - } - Err(err) => { - self.send_invalid_request_error( - request_id, - format!("failed to locate thread id {existing_thread_id}: {err}"), - ) - .await; - return None; - } - } + let params = StoreReadThreadParams { + thread_id: existing_thread_id, + include_archived: true, + include_history, + }; + self.thread_store.read_thread(params).await }; - match RolloutRecorder::get_rollout_history(&rollout_path).await { - Ok(initial_history) => Some(initial_history), + match result { + Ok(thread) => Some(thread), Err(err) => { - self.send_invalid_request_error( + self.outgoing + .send_error(request_id, thread_store_resume_read_error(err)) + .await; + None + } + } + } + + async fn stored_thread_to_initial_history( + &self, + request_id: ConnectionRequestId, + stored_thread: &StoredThread, + ) -> Option { + let thread_id = stored_thread.thread_id; + let history = match stored_thread.history.as_ref() { + Some(history) => history.items.clone(), + None => { + self.send_internal_error( request_id, - format!("failed to load rollout `{}`: {err}", rollout_path.display()), + format!("thread {thread_id} did not include persisted history"), ) .await; + return None; + } + }; + Some(InitialHistory::Resumed(ResumedHistory { + conversation_id: thread_id, + history, + rollout_path: stored_thread.rollout_path.clone(), + })) + } + + async fn stored_thread_to_api_thread( + &self, + stored_thread: StoredThread, + fallback_provider: &str, + include_turns: bool, + ) -> std::result::Result { + let (mut thread, history) = + thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd); + if include_turns && let Some(history) = history { + populate_thread_turns( + &mut thread, + ThreadTurnSource::HistoryItems(&history.items), + /*active_turn*/ None, + ) + .await?; + } + Ok(thread) + } + + async fn read_stored_thread_for_new_fork( + &self, + request_id: ConnectionRequestId, + thread_store: &dyn ThreadStore, + thread_id: ThreadId, + include_history: bool, + ) -> Option { + match thread_store + .read_thread(StoreReadThreadParams { + thread_id, + include_archived: true, + include_history, + }) + .await + { + Ok(thread) => Some(thread), + Err(err) => { + self.outgoing + .send_error(request_id, thread_store_resume_read_error(err)) + .await; None } } @@ -4982,20 +5042,42 @@ impl CodexMessageProcessor { thread: &CodexThread, thread_history: &InitialHistory, rollout_path: &Path, - persisted_resume_metadata: Option<&ThreadMetadata>, + resume_source_thread: Option, include_turns: bool, ) -> std::result::Result { let config_snapshot = thread.config_snapshot().await; let thread = match thread_history { InitialHistory::Resumed(resumed) => { - load_thread_summary_for_rollout( - &self.config, - resumed.conversation_id, - resumed.rollout_path.as_path(), - config_snapshot.model_provider_id.as_str(), - persisted_resume_metadata, - ) - .await + let fallback_provider = config_snapshot.model_provider_id.as_str(); + if let Some(mut stored_thread) = resume_source_thread { + stored_thread.history = None; + Ok(thread_from_stored_thread( + stored_thread, + fallback_provider, + &self.config.cwd, + ) + .0) + } else { + match self + .thread_store + .read_thread(StoreReadThreadParams { + thread_id: resumed.conversation_id, + include_archived: true, + include_history: false, + }) + .await + { + Ok(stored_thread) => Ok(thread_from_stored_thread( + stored_thread, + fallback_provider, + &self.config.cwd, + ) + .0), + Err(read_err) => { + Err(format!("failed to read thread from store: {read_err}")) + } + } + } } InitialHistory::Forked(items) => { let mut thread = build_thread_from_snapshot( @@ -5061,50 +5143,31 @@ impl CodexMessageProcessor { return; } - let (rollout_path, source_thread_id) = if let Some(path) = path { - (path, None) - } else { - let existing_thread_id = match ThreadId::from_string(&thread_id) { - Ok(id) => id, - Err(err) => { - self.send_invalid_request_error( - request_id, - format!("invalid thread id: {err}"), - ) - .await; - return; - } - }; - - match find_thread_path_by_id_str( - &self.config.codex_home, - &existing_thread_id.to_string(), + let Some(source_thread) = self + .read_stored_thread_for_resume( + request_id.clone(), + &thread_id, + path.as_ref(), + /*include_history*/ true, ) .await - { - Ok(Some(p)) => (p, Some(existing_thread_id)), - Ok(None) => { - self.send_invalid_request_error( - request_id, - format!("no rollout found for thread id {existing_thread_id}"), - ) - .await; - return; - } - Err(err) => { - self.send_invalid_request_error( - request_id, - format!("failed to locate thread id {existing_thread_id}: {err}"), - ) - .await; - return; - } - } + else { + return; }; - - let history_cwd = - read_history_cwd_from_state_db(&self.config, source_thread_id, rollout_path.as_path()) - .await; + let source_thread_id = source_thread.thread_id; + let Some(history_items) = source_thread + .history + .as_ref() + .map(|history| history.items.clone()) + else { + self.send_internal_error( + request_id, + format!("thread {source_thread_id} did not include persisted history"), + ) + .await; + return; + }; + let history_cwd = Some(source_thread.cwd.clone()); // Persist Windows sandbox mode. let mut cli_overrides = cli_overrides.unwrap_or_default(); @@ -5159,6 +5222,7 @@ impl CodexMessageProcessor { let fallback_model_provider = config.model_provider_id.clone(); let instruction_sources = Self::instruction_sources_from_config(&config).await; + let fork_thread_store = configured_thread_store(&config); let NewThread { thread_id, @@ -5167,10 +5231,14 @@ impl CodexMessageProcessor { .. } = match self .thread_manager - .fork_thread( + .fork_thread_from_history( ForkSnapshot::Interrupted, config, - rollout_path.clone(), + InitialHistory::Resumed(ResumedHistory { + conversation_id: source_thread_id, + history: history_items.clone(), + rollout_path: source_thread.rollout_path.clone(), + }), persist_extended_history, self.request_trace_context(&request_id).await, ) @@ -5182,7 +5250,7 @@ impl CodexMessageProcessor { CodexErr::Io(_) | CodexErr::Json(_) => { self.send_invalid_request_error( request_id, - format!("failed to load rollout `{}`: {err}", rollout_path.display()), + format!("failed to load thread {source_thread_id}: {err}"), ) .await; } @@ -5216,25 +5284,33 @@ impl CodexMessageProcessor { ); // Persistent forks materialize their own rollout immediately. Ephemeral forks stay - // pathless, so they rebuild their visible history from the copied source rollout instead. + // pathless, so they rebuild their visible history from the copied source history instead. let mut thread = if let Some(fork_rollout_path) = session_configured.rollout_path.as_ref() { - match read_summary_from_rollout( - fork_rollout_path.as_path(), - fallback_model_provider.as_str(), - ) - .await + let Some(stored_thread) = self + .read_stored_thread_for_new_fork( + request_id.clone(), + fork_thread_store.as_ref(), + thread_id, + include_turns, + ) + .await + else { + return; + }; + match self + .stored_thread_to_api_thread( + stored_thread, + fallback_model_provider.as_str(), + include_turns, + ) + .await { - Ok(summary) => { - let mut thread = summary_to_thread(summary, &self.config.cwd); - thread.forked_from_id = - forked_from_id_from_rollout(fork_rollout_path.as_path()).await; - thread - } - Err(err) => { + Ok(thread) => thread, + Err(message) => { self.send_internal_error( request_id, format!( - "failed to load rollout `{}` for thread {thread_id}: {err}", + "failed to load rollout `{}` for thread {thread_id}: {message}", fork_rollout_path.display() ), ) @@ -5247,30 +5323,8 @@ impl CodexMessageProcessor { // forked thread names do not inherit the source thread name let mut thread = build_thread_from_snapshot(thread_id, &config_snapshot, /*path*/ None); - let history_items = match read_rollout_items_from_rollout(rollout_path.as_path()).await - { - Ok(items) => items, - Err(err) => { - self.send_internal_error( - request_id, - format!( - "failed to load source rollout `{}` for thread {thread_id}: {err}", - rollout_path.display() - ), - ) - .await; - return; - } - }; thread.preview = preview_from_rollout_items(&history_items); - thread.forked_from_id = source_thread_id - .or_else(|| { - history_items.iter().find_map(|item| match item { - RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.id), - _ => None, - }) - }) - .map(|id| id.to_string()); + thread.forked_from_id = Some(source_thread_id.to_string()); if include_turns && let Err(message) = populate_thread_turns( &mut thread, @@ -5285,19 +5339,6 @@ impl CodexMessageProcessor { thread }; - if let Some(fork_rollout_path) = session_configured.rollout_path.as_ref() - && include_turns - && let Err(message) = populate_thread_turns( - &mut thread, - ThreadTurnSource::RolloutPath(fork_rollout_path.as_path()), - /*active_turn*/ None, - ) - .await - { - self.send_internal_error(request_id, message).await; - return; - } - self.thread_watch_manager .upsert_thread_silently(thread.clone()) .await; @@ -5346,11 +5387,10 @@ impl CodexMessageProcessor { { Some(turn_id) } else { - latest_token_usage_turn_id_from_rollout_path( - rollout_path.as_path(), + latest_token_usage_turn_id_from_rollout_items( + &history_items, token_usage_thread.turns.as_slice(), ) - .await }; // Mirror the resume contract for forks: the new thread is usable as soon // as the response arrives, so restored usage must follow immediately. @@ -5393,14 +5433,13 @@ impl CodexMessageProcessor { .as_any() .downcast_ref::() else { - let error = JSONRPCErrorError { - code: INVALID_REQUEST_ERROR_CODE, - message: - "rollout path queries are only supported with the local thread store" - .to_string(), - data: None, - }; - return self.outgoing.send_error(request_id, error).await; + self.send_invalid_request_error( + request_id, + "rollout path queries are only supported with the local thread store" + .to_string(), + ) + .await; + return; }; local_thread_store @@ -8812,7 +8851,7 @@ async fn handle_pending_thread_resume_request( if pending.include_turns && let Err(message) = populate_thread_turns( &mut thread, - ThreadTurnSource::RolloutPath(pending.rollout_path.as_path()), + ThreadTurnSource::HistoryItems(&pending.history_items), active_turn.as_ref(), ) .await @@ -8904,11 +8943,10 @@ async fn handle_pending_thread_resume_request( // Match cold resume: metadata-only resume should attach the listener without // paying the cost of turn reconstruction for historical usage replay. if let Some(token_usage_thread) = token_usage_thread { - let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_path( - pending.rollout_path.as_path(), + let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_items( + &pending.history_items, token_usage_thread.turns.as_slice(), - ) - .await; + ); // Rejoining a loaded thread has the same UI contract as a cold resume, but // uses the live conversation state instead of reconstructing a new session. send_thread_token_usage_update_to_connection( @@ -8927,7 +8965,6 @@ async fn handle_pending_thread_resume_request( } enum ThreadTurnSource<'a> { - RolloutPath(&'a Path), HistoryItems(&'a [RolloutItem]), } @@ -8937,18 +8974,6 @@ async fn populate_thread_turns( active_turn: Option<&Turn>, ) -> std::result::Result<(), String> { let mut turns = match turn_source { - ThreadTurnSource::RolloutPath(rollout_path) => { - read_rollout_items_from_rollout(rollout_path) - .await - .map(|items| build_turns_from_rollout_items(&items)) - .map_err(|err| { - format!( - "failed to load rollout `{}` for thread {}: {err}", - rollout_path.display(), - thread.id - ) - })? - } ThreadTurnSource::HistoryItems(items) => build_turns_from_rollout_items(items), }; if let Some(active_turn) = active_turn { @@ -9372,36 +9397,6 @@ fn validate_dynamic_tools(tools: &[ApiDynamicToolSpec]) -> Result<(), String> { Ok(()) } -async fn read_history_cwd_from_state_db( - config: &Config, - thread_id: Option, - rollout_path: &Path, -) -> Option { - if let Some(state_db_ctx) = get_state_db(config).await - && let Some(thread_id) = thread_id - && let Ok(Some(metadata)) = state_db_ctx.get_thread(thread_id).await - { - return Some(metadata.cwd); - } - - match read_session_meta_line(rollout_path).await { - Ok(meta_line) => Some(meta_line.meta.cwd), - Err(err) => { - let rollout_path = rollout_path.display(); - warn!("failed to read session metadata from rollout {rollout_path}: {err}"); - None - } - } -} - -async fn read_summary_from_state_db_by_thread_id( - config: &Config, - thread_id: ThreadId, -) -> Option { - let state_db_ctx = open_state_db_for_direct_thread_lookup(config).await; - read_summary_from_state_db_context_by_thread_id(state_db_ctx.as_ref(), thread_id).await -} - async fn read_summary_from_state_db_context_by_thread_id( state_db_ctx: Option<&StateDbHandle>, thread_id: ThreadId, @@ -9495,6 +9490,26 @@ fn thread_store_list_error(err: ThreadStoreError) -> JSONRPCErrorError { } } +fn thread_store_resume_read_error(err: ThreadStoreError) -> JSONRPCErrorError { + match err { + ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message, + data: None, + }, + ThreadStoreError::ThreadNotFound { thread_id } => JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("no rollout found for thread id {thread_id}"), + data: None, + }, + err => JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to read thread: {err}"), + data: None, + }, + } +} + fn conversation_summary_thread_id_read_error( conversation_id: ThreadId, err: ThreadStoreError, @@ -9914,45 +9929,6 @@ fn map_git_info(git_info: &CoreGitInfo) -> ConversationGitInfo { } } -async fn load_thread_summary_for_rollout( - config: &Config, - thread_id: ThreadId, - rollout_path: &Path, - fallback_provider: &str, - persisted_metadata: Option<&ThreadMetadata>, -) -> std::result::Result { - let mut thread = read_summary_from_rollout(rollout_path, fallback_provider) - .await - .map(|summary| summary_to_thread(summary, &config.cwd)) - .map_err(|err| { - format!( - "failed to load rollout `{}` for thread {thread_id}: {err}", - rollout_path.display() - ) - })?; - thread.forked_from_id = forked_from_id_from_rollout(rollout_path).await; - if let Some(persisted_metadata) = persisted_metadata { - merge_mutable_thread_metadata( - &mut thread, - summary_to_thread( - summary_from_thread_metadata(persisted_metadata), - &config.cwd, - ), - ); - } else if let Some(summary) = read_summary_from_state_db_by_thread_id(config, thread_id).await { - merge_mutable_thread_metadata(&mut thread, summary_to_thread(summary, &config.cwd)); - } - let title = if let Some(metadata) = persisted_metadata { - non_empty_title(metadata) - } else { - title_from_state_db(config, thread_id).await - }; - if let Some(title) = title { - set_thread_name_from_title(&mut thread, title); - } - Ok(thread) -} - async fn forked_from_id_from_rollout(path: &Path) -> Option { read_session_meta_line(path) .await @@ -9961,10 +9937,6 @@ async fn forked_from_id_from_rollout(path: &Path) -> Option { .map(|thread_id| thread_id.to_string()) } -fn merge_mutable_thread_metadata(thread: &mut Thread, persisted_thread: Thread) { - thread.git_info = persisted_thread.git_info; -} - fn preview_from_rollout_items(items: &[RolloutItem]) -> String { items .iter() diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 77b6defab..d4347933e 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -8,10 +8,10 @@ use codex_core::CodexThread; use codex_core::ThreadConfigSnapshot; use codex_protocol::ThreadId; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::RolloutItem; use codex_utils_absolute_path::AbsolutePathBuf; use std::collections::HashMap; use std::collections::HashSet; -use std::path::PathBuf; use std::sync::Arc; use std::sync::Weak; use tokio::sync::Mutex; @@ -27,7 +27,7 @@ type PendingInterruptQueue = Vec<( pub(crate) struct PendingThreadResumeRequest { pub(crate) request_id: ConnectionRequestId, - pub(crate) rollout_path: PathBuf, + pub(crate) history_items: Vec, pub(crate) config_snapshot: ThreadConfigSnapshot, pub(crate) instruction_sources: Vec, pub(crate) thread_summary: codex_app_server_protocol::Thread, diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index 7274daaa5..6c43ebd62 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -32,6 +32,7 @@ use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; use std::path::Path; +use std::path::PathBuf; use tempfile::TempDir; use tokio::time::timeout; use wiremock::Mock; @@ -49,6 +50,7 @@ use super::analytics::wait_for_analytics_payload; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25); #[cfg(not(windows))] const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +const INTERNAL_ERROR_CODE: i64 = -32603; #[tokio::test] async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { @@ -195,6 +197,88 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_fork_can_load_source_by_path() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let preview = "Saved user message"; + let conversation_id = create_fake_rollout( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + preview, + Some("mock_provider"), + /*git_info*/ None, + )?; + let original_path = codex_home + .path() + .join("sessions") + .join("2025") + .join("01") + .join("05") + .join(format!( + "rollout-2025-01-05T12-00-00-{conversation_id}.jsonl" + )); + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: "not-a-valid-thread-id".to_string(), + path: Some(original_path), + ..Default::default() + }) + .await?; + let fork_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(fork_id)), + ) + .await??; + let ThreadForkResponse { thread, .. } = to_response::(fork_resp)?; + + assert_ne!(thread.id, conversation_id); + assert_eq!(thread.forked_from_id, Some(conversation_id)); + assert_eq!(thread.preview, preview); + assert_eq!(thread.model_provider, "mock_provider"); + assert_eq!(thread.turns.len(), 1, "expected copied fork history"); + + Ok(()) +} + +#[tokio::test] +async fn thread_fork_by_path_uses_remote_thread_store_error() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml_with_remote_thread_store(codex_home.path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: "not-a-valid-thread-id".to_string(), + path: Some(PathBuf::from("sessions/2025/01/05/rollout.jsonl")), + ..Default::default() + }) + .await?; + let fork_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(fork_id)), + ) + .await??; + + assert_eq!(fork_err.error.code, INTERNAL_ERROR_CODE); + assert_eq!( + fork_err.error.message, + "failed to read thread: thread-store internal error: remote thread store does not support read_thread_by_rollout_path" + ); + + Ok(()) +} + #[tokio::test] async fn thread_fork_emits_restored_token_usage_before_next_turn() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; @@ -678,6 +762,33 @@ stream_max_retries = 0 ) } +fn create_config_toml_with_remote_thread_store( + codex_home: &Path, + server_uri: &str, +) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "read-only" +experimental_thread_store_endpoint = "http://127.0.0.1:1" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} + fn create_config_toml_with_chatgpt_base_url( codex_home: &Path, server_uri: &str, 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 c86f88825..e450dd50d 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -2,6 +2,7 @@ use anyhow::Result; use app_test_support::ChatGptAuthFixture; use app_test_support::McpProcess; use app_test_support::create_apply_patch_sse_response; +use app_test_support::create_fake_rollout; use app_test_support::create_fake_rollout_with_text_elements; use app_test_support::create_fake_rollout_with_token_usage; use app_test_support::create_final_assistant_message_sse_response; @@ -87,6 +88,7 @@ use super::analytics::wait_for_analytics_payload; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25); #[cfg(not(windows))] const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +const INTERNAL_ERROR_CODE: i64 = -32603; const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals."; async fn wait_for_responses_request_count( @@ -324,6 +326,37 @@ async fn thread_resume_can_skip_turns_for_metadata_only_resume() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_resume_by_path_uses_remote_thread_store_error() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml_with_remote_thread_store(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: "ignored-when-path-is-present".to_string(), + path: Some(PathBuf::from("sessions/2025/01/05/rollout.jsonl")), + ..Default::default() + }) + .await?; + let resume_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(resume_id)), + ) + .await??; + + assert_eq!(resume_err.error.code, INTERNAL_ERROR_CODE); + assert_eq!( + resume_err.error.message, + "failed to read thread: thread-store internal error: remote thread store does not support read_thread_by_rollout_path" + ); + + Ok(()) +} + #[tokio::test] async fn thread_resume_emits_restored_token_usage_before_next_turn() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; @@ -978,6 +1011,22 @@ async fn thread_resume_without_overrides_does_not_change_updated_at_or_mtime() - let mut mcp = McpProcess::new(codex_home.path()).await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + let read_id = mcp + .send_thread_read_request(ThreadReadParams { + thread_id: thread_id.clone(), + include_turns: false, + }) + .await?; + let read_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(read_id)), + ) + .await??; + let ThreadReadResponse { + thread: before_resume, + .. + } = to_response::(read_resp)?; + let resume_id = mcp .send_thread_resume_request(ThreadResumeParams { thread_id: thread_id.clone(), @@ -991,7 +1040,7 @@ async fn thread_resume_without_overrides_does_not_change_updated_at_or_mtime() - .await??; let ThreadResumeResponse { thread, .. } = to_response::(resume_resp)?; - assert_eq!(thread.updated_at, rollout.expected_updated_at); + assert_eq!(thread.updated_at, before_resume.updated_at); assert_eq!(thread.status, ThreadStatus::Idle); let after_modified = std::fs::metadata(&rollout.rollout_file_path)?.modified()?; @@ -1842,13 +1891,11 @@ async fn thread_resume_with_overrides_defers_updated_at_until_turn_start() -> Re mut mcp, thread_id, rollout_file_path, + updated_at, } = start_materialized_thread_and_restart(codex_home.path(), "materialize").await?; let expected_updated_at_rfc3339 = "2025-01-07T00:00:00Z"; 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(); let resume_id = mcp .send_thread_resume_request(ThreadResumeParams { @@ -1867,7 +1914,7 @@ async fn thread_resume_with_overrides_defers_updated_at_until_turn_start() -> Re .. } = to_response::(resume_resp)?; - assert_eq!(resumed_thread.updated_at, expected_updated_at); + assert_eq!(resumed_thread.updated_at, updated_at); assert_eq!(resumed_thread.status, ThreadStatus::Idle); let after_resume_modified = std::fs::metadata(&rollout_file_path)?.modified()?; @@ -2098,6 +2145,49 @@ async fn thread_resume_prefers_path_over_thread_id() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_resume_can_load_source_by_external_path() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + let external_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + let thread_id = create_fake_rollout( + external_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + "external path history", + Some("mock_provider"), + /*git_info*/ None, + )?; + let thread_path = rollout_path(external_home.path(), "2025-01-05T12-00-00", &thread_id); + + 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: "not-a-valid-thread-id".to_string(), + path: Some(thread_path.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: resumed, .. + } = to_response::(resume_resp)?; + let expected_thread_path = std::fs::canonicalize(&thread_path)?; + assert_eq!(resumed.id, thread_id); + assert_eq!(resumed.path, Some(expected_thread_path)); + assert_eq!(resumed.preview, "external path history"); + assert_eq!(resumed.status, ThreadStatus::Idle); + + Ok(()) +} + #[tokio::test] async fn thread_resume_supports_history_and_overrides() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; @@ -2151,6 +2241,7 @@ struct RestartedThreadFixture { mcp: McpProcess, thread_id: String, rollout_file_path: PathBuf, + updated_at: i64, } async fn start_materialized_thread_and_restart( @@ -2194,10 +2285,24 @@ async fn start_materialized_thread_and_restart( ) .await??; + let read_id = first_mcp + .send_thread_read_request(ThreadReadParams { + thread_id: thread.id.clone(), + include_turns: false, + }) + .await?; + let read_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + first_mcp.read_stream_until_response_message(RequestId::Integer(read_id)), + ) + .await??; + let ThreadReadResponse { thread, .. } = to_response::(read_resp)?; + let thread_id = thread.id; let rollout_file_path = thread .path .ok_or_else(|| anyhow::anyhow!("thread path missing from thread/start response"))?; + let updated_at = thread.updated_at; drop(first_mcp); @@ -2208,6 +2313,7 @@ async fn start_materialized_thread_and_restart( mcp: second_mcp, thread_id, rollout_file_path: rollout_file_path.to_path_buf(), + updated_at, }) } @@ -2357,6 +2463,37 @@ stream_max_retries = 0 ) } +fn create_config_toml_with_remote_thread_store( + codex_home: &std::path::Path, + server_uri: &str, +) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + format!( + r#" +model = "gpt-5.3-codex" +approval_policy = "never" +sandbox_mode = "read-only" +experimental_thread_store_endpoint = "http://127.0.0.1:1" + +model_provider = "mock_provider" + +[features] +personality = true +general_analytics = true + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} + fn create_config_toml_with_chatgpt_base_url( codex_home: &std::path::Path, server_uri: &str, @@ -2443,7 +2580,6 @@ 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 { @@ -2465,14 +2601,9 @@ fn setup_rollout_fixture(codex_home: &Path, server_uri: &str) -> Result( + &self, + snapshot: S, + config: Config, + history: InitialHistory, + persist_extended_history: bool, + parent_trace: Option, + ) -> CodexResult + where + S: Into, + { + self.fork_thread_with_initial_history( + snapshot.into(), + config, + history, + persist_extended_history, + parent_trace, + ) + .await + } + + async fn fork_thread_with_initial_history( + &self, + snapshot: ForkSnapshot, + config: Config, + history: InitialHistory, + persist_extended_history: bool, + parent_trace: Option, + ) -> CodexResult { let interrupted_marker = InterruptedTurnHistoryMarker::from_config(&config); - let history = match snapshot { - ForkSnapshot::TruncateBeforeNthUserMessage(nth_user_message) => { - truncate_before_nth_user_message(history, nth_user_message, &snapshot_state) - } - ForkSnapshot::Interrupted => { - let history = match history { - InitialHistory::New => InitialHistory::New, - InitialHistory::Cleared => InitialHistory::Cleared, - InitialHistory::Forked(history) => InitialHistory::Forked(history), - InitialHistory::Resumed(resumed) => InitialHistory::Forked(resumed.history), - }; - if snapshot_state.ends_mid_turn { - append_interrupted_boundary( - history, - snapshot_state.active_turn_id, - interrupted_marker, - ) - } else { - history - } - } - }; + let history = fork_history_from_snapshot(snapshot, history, interrupted_marker); let thread_store = configured_thread_store(&config); let environments = default_thread_environment_selections( self.state.environment_manager.as_ref(), @@ -1228,6 +1246,36 @@ fn snapshot_turn_state(history: &InitialHistory) -> SnapshotTurnState { } } +fn fork_history_from_snapshot( + snapshot: ForkSnapshot, + history: InitialHistory, + interrupted_marker: InterruptedTurnHistoryMarker, +) -> InitialHistory { + let snapshot_state = snapshot_turn_state(&history); + match snapshot { + ForkSnapshot::TruncateBeforeNthUserMessage(nth_user_message) => { + truncate_before_nth_user_message(history, nth_user_message, &snapshot_state) + } + ForkSnapshot::Interrupted => { + let history = match history { + InitialHistory::New => InitialHistory::New, + InitialHistory::Cleared => InitialHistory::Cleared, + InitialHistory::Forked(history) => InitialHistory::Forked(history), + InitialHistory::Resumed(resumed) => InitialHistory::Forked(resumed.history), + }; + if snapshot_state.ends_mid_turn { + append_interrupted_boundary( + history, + snapshot_state.active_turn_id, + interrupted_marker, + ) + } else { + history + } + } + } +} + /// Append the same persisted interrupt boundary used by the live interrupt path /// to an existing fork snapshot after the source thread has been confirmed to /// be mid-turn. diff --git a/codex-rs/core/tests/suite/fork_thread.rs b/codex-rs/core/tests/suite/fork_thread.rs index bcb7864cf..7648e97f8 100644 --- a/codex-rs/core/tests/suite/fork_thread.rs +++ b/codex-rs/core/tests/suite/fork_thread.rs @@ -3,7 +3,9 @@ use codex_core::NewThread; use codex_core::parse_turn_item; use codex_protocol::items::TurnItem; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::Op; +use codex_protocol::protocol::ResumedHistory; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; use codex_protocol::user_input::UserInput; @@ -66,27 +68,9 @@ async fn fork_thread_twice_drops_to_first_message() { // GetHistory flushes before returning the path; no wait needed. - // Helper: read rollout items (excluding SessionMeta) from a JSONL path. - let read_items = |p: &std::path::Path| -> Vec { - let text = std::fs::read_to_string(p).expect("read rollout file"); - let mut items: Vec = Vec::new(); - for line in text.lines() { - if line.trim().is_empty() { - continue; - } - let v: serde_json::Value = serde_json::from_str(line).expect("jsonl line"); - let rl: RolloutLine = serde_json::from_value(v).expect("rollout line"); - match rl.item { - RolloutItem::SessionMeta(_) => {} - other => items.push(other), - } - } - items - }; - // Compute expected prefixes after each fork by truncating base rollout // strictly before the nth user input (0-based). - let base_items = read_items(&base_path); + let base_items = read_rollout_items(&base_path); let find_user_input_positions = |items: &[RolloutItem]| -> Vec { let mut pos = Vec::new(); for (i, it) in items.iter().enumerate() { @@ -126,7 +110,7 @@ async fn fork_thread_twice_drops_to_first_message() { let fork1_path = codex_fork1.rollout_path().expect("rollout path"); // GetHistory on fork1 flushed; the file is ready. - let fork1_items = read_items(&fork1_path); + let fork1_items = read_rollout_items(&fork1_path); pretty_assertions::assert_eq!( serde_json::to_value(&fork1_items).unwrap(), serde_json::to_value(&expected_after_first).unwrap() @@ -149,16 +133,114 @@ async fn fork_thread_twice_drops_to_first_message() { let fork2_path = codex_fork2.rollout_path().expect("rollout path"); // GetHistory on fork2 flushed; the file is ready. - let fork1_items = read_items(&fork1_path); + let fork1_items = read_rollout_items(&fork1_path); let fork1_user_inputs = find_user_input_positions(&fork1_items); let cut_last_on_fork1 = fork1_user_inputs .get(fork1_user_inputs.len().saturating_sub(1)) .copied() .unwrap_or(0); let expected_after_second: Vec = fork1_items[..cut_last_on_fork1].to_vec(); - let fork2_items = read_items(&fork2_path); + let fork2_items = read_rollout_items(&fork2_path); pretty_assertions::assert_eq!( serde_json::to_value(&fork2_items).unwrap(), serde_json::to_value(&expected_after_second).unwrap() ); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn fork_thread_from_history_does_not_require_source_rollout_path() { + skip_if_no_network!(); + + let server = MockServer::start().await; + let sse = sse(vec![ev_response_created("resp"), ev_completed("resp")]); + Mock::given(method("POST")) + .and(path("/v1/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_raw(sse, "text/event-stream"), + ) + .expect(1) + .mount(&server) + .await; + + let mut builder = test_codex(); + let test = builder.build(&server).await.expect("create conversation"); + let codex = test.codex.clone(); + let thread_manager = test.thread_manager.clone(); + + codex + .submit(Op::UserInput { + environments: None, + items: vec![UserInput::Text { + text: "fork me from stored history".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + }) + .await + .unwrap(); + let _ = wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + + let source_path = codex.rollout_path().expect("source rollout path"); + let source_items = read_rollout_items(&source_path); + let NewThread { + thread: forked_thread, + .. + } = thread_manager + .fork_thread_from_history( + ForkSnapshot::Interrupted, + test.config, + InitialHistory::Resumed(ResumedHistory { + conversation_id: test.session_configured.session_id, + history: source_items.clone(), + rollout_path: None, + }), + /*persist_extended_history*/ false, + /*parent_trace*/ None, + ) + .await + .expect("fork from stored history"); + + let forked_path = forked_thread.rollout_path().expect("forked rollout path"); + let forked_items = read_rollout_items(&forked_path); + let forked_items = forked_items + .iter() + .map(|item| serde_json::to_value(item).unwrap()) + .collect::>(); + let source_items = source_items + .iter() + .map(|item| serde_json::to_value(item).unwrap()) + .collect::>(); + assert!( + forked_items.starts_with(&source_items), + "forked history should start with the supplied source history" + ); +} + +fn read_rollout_items(path: &std::path::Path) -> Vec { + let text = match std::fs::read_to_string(path) { + Ok(text) => text, + Err(err) => panic!("failed to read rollout file {}: {err}", path.display()), + }; + let mut items: Vec = Vec::new(); + for line in text.lines() { + if line.trim().is_empty() { + continue; + } + let v: serde_json::Value = match serde_json::from_str(line) { + Ok(value) => value, + Err(err) => panic!("failed to parse rollout JSON line `{line}`: {err}"), + }; + let rl: RolloutLine = match serde_json::from_value(v) { + Ok(line) => line, + Err(err) => panic!("failed to parse rollout line `{line}`: {err}"), + }; + match rl.item { + RolloutItem::SessionMeta(_) => {} + other => items.push(other), + } + } + items +} diff --git a/codex-rs/core/tests/suite/resume_warning.rs b/codex-rs/core/tests/suite/resume_warning.rs index e1e9220fa..ee3c7bbf3 100644 --- a/codex-rs/core/tests/suite/resume_warning.rs +++ b/codex-rs/core/tests/suite/resume_warning.rs @@ -74,7 +74,7 @@ fn resume_history( time_to_first_token_ms: None, })), ], - rollout_path: rollout_path.to_path_buf(), + rollout_path: Some(rollout_path.to_path_buf()), }) } diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index f2219cdcf..a79aa1f01 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -2640,7 +2640,7 @@ pub struct ConversationPathResponseEvent { pub struct ResumedHistory { pub conversation_id: ThreadId, pub history: Vec, - pub rollout_path: PathBuf, + pub rollout_path: Option, } #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index cd93b6d55..7b3037c4a 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -913,7 +913,7 @@ impl RolloutRecorder { Ok(InitialHistory::Resumed(ResumedHistory { conversation_id, history: items, - rollout_path: path.to_path_buf(), + rollout_path: Some(path.to_path_buf()), })) } diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 15ec238fa..c8a083e1c 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -25,6 +25,7 @@ pub use types::GitInfoPatch; pub use types::ListThreadsParams; pub use types::LoadThreadHistoryParams; pub use types::OptionalStringPatch; +pub use types::ReadThreadByRolloutPathParams; pub use types::ReadThreadParams; pub use types::ResumeThreadParams; pub use types::SortDirection; diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index e5f73ff83..a246a41ae 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -27,6 +27,7 @@ use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::ListThreadsParams; use crate::LoadThreadHistoryParams; +use crate::ReadThreadByRolloutPathParams; use crate::ReadThreadParams; use crate::ResumeThreadParams; use crate::StoredThread; @@ -207,6 +208,19 @@ impl ThreadStore for LocalThreadStore { read_thread::read_thread(self, params).await } + async fn read_thread_by_rollout_path( + &self, + params: ReadThreadByRolloutPathParams, + ) -> ThreadStoreResult { + read_thread::read_thread_by_rollout_path( + self, + params.rollout_path, + params.include_archived, + params.include_history, + ) + .await + } + async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult { list_threads::list_threads(self, params).await } @@ -508,6 +522,51 @@ mod tests { })); } + #[tokio::test] + async fn read_thread_by_rollout_path_includes_history() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let thread_id = ThreadId::default(); + + store + .create_thread(create_thread_params(thread_id)) + .await + .expect("create thread"); + store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![user_message_item("path read")], + }) + .await + .expect("append item"); + store.flush_thread(thread_id).await.expect("flush thread"); + let rollout_path = store + .live_rollout_path(thread_id) + .await + .expect("load rollout path"); + + let thread = store + .read_thread_by_rollout_path( + rollout_path, + /*include_archived*/ true, + /*include_history*/ true, + ) + .await + .expect("read thread by rollout path"); + + assert_eq!(thread.thread_id, thread_id); + assert_eq!( + thread + .history + .expect("history") + .items + .into_iter() + .filter(|item| matches!(item, RolloutItem::EventMsg(EventMsg::UserMessage(_)))) + .count(), + 1 + ); + } + fn create_thread_params(thread_id: ThreadId) -> CreateThreadParams { CreateThreadParams { thread_id, diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs index 751a94015..7bbc7bb3d 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -29,6 +29,13 @@ pub(super) async fn read_thread( let thread_id = params.thread_id; if let Some(metadata) = read_sqlite_metadata(store, thread_id).await && (params.include_archived || metadata.archived_at.is_none()) + && (!params.include_history + || sqlite_rollout_path_can_load_history_for_thread( + store, + &metadata.rollout_path, + thread_id, + ) + .await) { let mut thread = stored_thread_from_sqlite_metadata(store, metadata).await; attach_history_if_requested(&mut thread, params.include_history).await?; @@ -46,6 +53,22 @@ pub(super) async fn read_thread( Ok(thread) } +async fn sqlite_rollout_path_can_load_history_for_thread( + store: &LocalThreadStore, + path: &std::path::Path, + thread_id: codex_protocol::ThreadId, +) -> bool { + if !tokio::fs::try_exists(path).await.unwrap_or(false) { + return false; + } + // SQLite metadata can outlive a moved/recreated rollout path. When history is + // requested, verify the path still resolves to the requested thread before + // trusting it as the source replay. + read_thread_from_rollout_path(store, path.to_path_buf()) + .await + .is_ok_and(|thread| thread.thread_id == thread_id) +} + pub(super) async fn read_thread_by_rollout_path( store: &LocalThreadStore, rollout_path: std::path::PathBuf, @@ -640,6 +663,102 @@ mod tests { assert_eq!(history.items.len(), 1); } + #[tokio::test] + async fn read_thread_falls_back_to_rollout_search_when_sqlite_path_is_stale() { + let home = TempDir::new().expect("temp dir"); + let external = TempDir::new().expect("external temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config.clone()); + let uuid = Uuid::from_u128(220); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let rollout_path = + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + let stale_path = external.path().join("missing-rollout.jsonl"); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let mut builder = ThreadMetadataBuilder::new( + thread_id, + stale_path.clone(), + Utc::now(), + SessionSource::Cli, + ); + builder.model_provider = Some("stale-sqlite-provider".to_string()); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.first_user_message = Some("stale sqlite preview".to_string()); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: true, + include_history: true, + }) + .await + .expect("read thread"); + + assert_eq!(thread.thread_id, thread_id); + assert_eq!(thread.rollout_path, Some(rollout_path)); + assert_eq!(thread.preview, "Hello from user"); + assert_eq!(thread.model_provider, config.model_provider_id); + let history = thread.history.expect("history should load"); + assert_eq!(history.thread_id, thread_id); + assert_eq!(history.items.len(), 2); + } + + #[tokio::test] + async fn read_thread_falls_back_when_sqlite_path_points_to_another_thread() { + let home = TempDir::new().expect("temp dir"); + let external = TempDir::new().expect("external temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config.clone()); + let uuid = Uuid::from_u128(221); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let rollout_path = + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + let other_uuid = Uuid::from_u128(222); + let stale_path = write_session_file(external.path(), "2025-01-04T12-00-00", other_uuid) + .expect("other session file"); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let mut builder = + ThreadMetadataBuilder::new(thread_id, stale_path, Utc::now(), SessionSource::Cli); + builder.model_provider = Some("wrong-sqlite-provider".to_string()); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.first_user_message = Some("wrong sqlite preview".to_string()); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: true, + include_history: true, + }) + .await + .expect("read thread"); + + assert_eq!(thread.thread_id, thread_id); + assert_eq!(thread.rollout_path, Some(rollout_path)); + assert_eq!(thread.preview, "Hello from user"); + assert_eq!(thread.model_provider, config.model_provider_id); + let history = thread.history.expect("history should load"); + assert_eq!(history.thread_id, thread_id); + assert_eq!(history.items.len(), 2); + } + #[tokio::test] async fn read_thread_uses_session_meta_for_rollout_without_user_preview_or_sqlite_metadata() { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/thread-store/src/remote/mod.rs b/codex-rs/thread-store/src/remote/mod.rs index 48fef2249..a3befac4e 100644 --- a/codex-rs/thread-store/src/remote/mod.rs +++ b/codex-rs/thread-store/src/remote/mod.rs @@ -9,6 +9,7 @@ use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::ListThreadsParams; use crate::LoadThreadHistoryParams; +use crate::ReadThreadByRolloutPathParams; use crate::ReadThreadParams; use crate::ResumeThreadParams; use crate::StoredThread; @@ -25,6 +26,10 @@ mod proto; /// gRPC-backed [`ThreadStore`] implementation for deployments whose durable thread data lives /// outside the app-server process. +/// +/// This store is still a work in progress: app-server code should call the generic +/// [`ThreadStore`] methods, and unsupported remote operations will return explicit +/// `not_implemented` errors until the remote API catches up. #[derive(Clone, Debug)] pub struct RemoteThreadStore { endpoint: String, @@ -187,6 +192,15 @@ impl ThreadStore for RemoteThreadStore { helpers::stored_thread_from_proto(thread) } + async fn read_thread_by_rollout_path( + &self, + _params: ReadThreadByRolloutPathParams, + ) -> ThreadStoreResult { + Err(ThreadStoreError::Internal { + message: "remote thread store does not support read_thread_by_rollout_path".to_string(), + }) + } + async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult { list_threads::list_threads(self, params).await } diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index def8bf082..238e56aa9 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -7,6 +7,7 @@ use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::ListThreadsParams; use crate::LoadThreadHistoryParams; +use crate::ReadThreadByRolloutPathParams; use crate::ReadThreadParams; use crate::ResumeThreadParams; use crate::StoredThread; @@ -18,8 +19,7 @@ use crate::UpdateThreadMetadataParams; /// Storage-neutral thread persistence boundary. #[async_trait] pub trait ThreadStore: Any + Send + Sync { - /// Return this store as [`Any`] so callers at API boundaries can reject requests that only - /// make sense for a concrete store implementation. + /// Return this store as [`Any`] for implementation-owned escape hatches. fn as_any(&self) -> &dyn Any; /// Creates a new live thread. @@ -56,6 +56,14 @@ pub trait ThreadStore: Any + Send + Sync { /// Reads a thread summary and optionally its persisted history. async fn read_thread(&self, params: ReadThreadParams) -> ThreadStoreResult; + /// Reads a rollout-backed thread by path when the store supports path-addressed lookups. + /// + /// Deprecated: new callers should use [`ThreadStore::read_thread`] instead. + async fn read_thread_by_rollout_path( + &self, + params: ReadThreadByRolloutPathParams, + ) -> ThreadStoreResult; + /// Lists stored threads matching the supplied filters. async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult; diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 537b09320..f019bcb29 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -96,6 +96,17 @@ pub struct ReadThreadParams { pub include_history: bool, } +/// Parameters for reading a local rollout-backed thread by path. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ReadThreadByRolloutPathParams { + /// Local rollout JSONL path to read. + pub rollout_path: PathBuf, + /// Whether archived threads are eligible. + pub include_archived: bool, + /// Whether persisted rollout items should be included in the response. + pub include_history: bool, +} + /// The sort key to use when listing stored threads. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub enum ThreadSortKey {