diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index a7a1f6ce81..da88d0e067 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -438,8 +438,8 @@ use codex_protocol::protocol::SessionConfiguredEvent; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::TurnEnvironmentSelection; use codex_protocol::protocol::TurnEnvironmentSelections; -use codex_protocol::protocol::USER_MESSAGE_BEGIN; use codex_protocol::protocol::W3cTraceContext; +use codex_protocol::protocol::strip_user_message_prefix; use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS; use codex_protocol::user_input::UserInput as CoreInputItem; use codex_rmcp_client::perform_oauth_login_return_url_with_http_client; @@ -625,10 +625,10 @@ pub(crate) use self::thread_summary::summary_to_thread; pub(crate) use self::thread_summary::thread_settings_from_config_snapshot; pub(crate) use self::thread_summary::thread_settings_from_core_snapshot; -pub(crate) fn build_api_turns_from_rollout_items(items: &[RolloutItem]) -> Vec { +pub(crate) fn build_legacy_api_turns_from_rollout_items(items: &[RolloutItem]) -> Vec { let mut builder = ThreadHistoryBuilder::new(); for item in items { - if is_persisted_rollout_item(item) { + if is_persisted_rollout_item(item, codex_protocol::protocol::ThreadHistoryMode::Legacy) { builder.handle_rollout_item(item); } } diff --git a/codex-rs/app-server/src/request_processors/external_agent_session_import.rs b/codex-rs/app-server/src/request_processors/external_agent_session_import.rs index e574b7a64b..6ec89401ec 100644 --- a/codex-rs/app-server/src/request_processors/external_agent_session_import.rs +++ b/codex-rs/app-server/src/request_processors/external_agent_session_import.rs @@ -231,7 +231,7 @@ impl ExternalAgentSessionImporter { memory_mode, }, }; - rollout_items.retain(is_persisted_rollout_item); + rollout_items.retain(|item| is_persisted_rollout_item(item, ThreadHistoryMode::Legacy)); let title = title .as_deref() .and_then(codex_core::util::normalize_thread_name); diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 784cd51add..0fc9194210 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -738,7 +738,7 @@ pub(crate) fn populate_thread_turns_from_history( items: &[RolloutItem], active_turn: Option<&Turn>, ) { - let mut turns = build_api_turns_from_rollout_items(items); + let mut turns = build_legacy_api_turns_from_rollout_items(items); if let Some(active_turn) = active_turn { merge_turn_history_with_active_turn(&mut turns, active_turn.clone()); } diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 88c8625642..fc6a3281b4 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2339,7 +2339,7 @@ impl ThreadRequestProcessor { let (mut thread, history) = thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd); if include_turns && let Some(history) = history { - thread.turns = build_api_turns_from_rollout_items(&history.items); + thread.turns = build_legacy_api_turns_from_rollout_items(&history.items); } Ok(Some(thread)) } @@ -2408,7 +2408,7 @@ impl ThreadRequestProcessor { .load_history(/*include_archived*/ true) .await .map_err(|err| thread_read_history_load_error(thread_id, err))?; - thread.turns = build_api_turns_from_rollout_items(&history.items); + thread.turns = build_legacy_api_turns_from_rollout_items(&history.items); } Ok(()) @@ -4107,7 +4107,7 @@ fn reconstruct_thread_turns_for_turns_list( || active_turn .as_ref() .is_some_and(|turn| matches!(turn.status, TurnStatus::InProgress)); - let mut turns = build_api_turns_from_rollout_items(items); + let mut turns = build_legacy_api_turns_from_rollout_items(items); normalize_thread_turns_status(&mut turns, loaded_status, has_live_in_progress_turn); if let Some(active_turn) = active_turn { merge_turn_history_with_active_turn(&mut turns, active_turn); @@ -4480,10 +4480,7 @@ fn preview_from_rollout_items(items: &[RolloutItem]) -> String { }, _ => None, }) - .map(|preview| match preview.find(USER_MESSAGE_BEGIN) { - Some(idx) => preview[idx + USER_MESSAGE_BEGIN.len()..].trim().to_string(), - None => preview, - }) + .map(|preview| strip_user_message_prefix(preview.as_str()).to_string()) .unwrap_or_default() } diff --git a/codex-rs/app-server/src/request_processors/thread_summary.rs b/codex-rs/app-server/src/request_processors/thread_summary.rs index 37c77a198b..139da67bca 100644 --- a/codex-rs/app-server/src/request_processors/thread_summary.rs +++ b/codex-rs/app-server/src/request_processors/thread_summary.rs @@ -97,10 +97,7 @@ fn extract_conversation_summary( _ => None, })?; - let preview = match preview.find(USER_MESSAGE_BEGIN) { - Some(idx) => preview[idx + USER_MESSAGE_BEGIN.len()..].trim(), - None => preview.as_str(), - }; + let preview = strip_user_message_prefix(preview.as_str()); let timestamp = if session_meta.timestamp.is_empty() { None diff --git a/codex-rs/app-server/src/request_processors/thread_summary_tests.rs b/codex-rs/app-server/src/request_processors/thread_summary_tests.rs index 8bead5529b..cca4b5e537 100644 --- a/codex-rs/app-server/src/request_processors/thread_summary_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_summary_tests.rs @@ -1,6 +1,7 @@ use super::*; use anyhow::Result; +use codex_protocol::protocol::USER_MESSAGE_BEGIN; use pretty_assertions::assert_eq; use serde_json::json; use std::path::PathBuf; diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 62212e9a16..9339413b33 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1379,7 +1379,7 @@ impl Session { } InitialHistory::Forked(mut rollout_items) => { let turn_context = self.new_default_turn().await; - if turn_context.config.features.enabled(Feature::ItemIds) { + if turn_context.item_ids_enabled() { for rollout_item in &mut rollout_items { if let RolloutItem::ResponseItem(response_item) = rollout_item { Self::assign_missing_response_item_id(response_item); @@ -1962,7 +1962,7 @@ impl Session { } async fn send_event_raw_with_persistence(&self, event: Event, persist: bool) { - // Persist the event into rollout storage (the store filters as needed). + // Persist the event into rollout storage; the store applies its persistence policy. if persist { let rollout_items = vec![RolloutItem::EventMsg(event.msg.clone())]; self.persist_rollout_items(&rollout_items).await; @@ -2776,7 +2776,7 @@ impl Session { for item in items.to_mut() { item.set_turn_id_if_missing(&turn_context.sub_id); } - if turn_context.config.features.enabled(Feature::ItemIds) { + if turn_context.item_ids_enabled() { Self::assign_missing_response_item_ids(items) } else { items @@ -3035,7 +3035,7 @@ impl Session { world_state_baseline: Option>, compacted_item: CompactedItem, ) { - let items = if turn_context.config.features.enabled(Feature::ItemIds) { + let items = if turn_context.item_ids_enabled() { Self::assign_missing_response_item_ids(Cow::Owned(items)).into_owned() } else { items diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index 3379a340fb..6b68447b92 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -120,6 +120,7 @@ pub(super) async fn spawn_review_thread( reasoning_effort, reasoning_summary, session_source, + history_mode: parent_turn_context.history_mode, parent_thread_id: parent_turn_context.parent_thread_id, originator: parent_turn_context.originator.clone(), environments: parent_turn_context.environments.clone(), diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index d9b3e81db7..32be4813ec 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -634,7 +634,12 @@ impl Session { }, }, }; - LiveThread::resume(Arc::clone(&thread_store), params).await? + LiveThread::resume( + Arc::clone(&thread_store), + session_configuration.history_mode, + params, + ) + .await? } }; Ok(Some(live_thread)) @@ -1110,7 +1115,11 @@ impl Session { config.features.enabled(Feature::EnableRequestCompression), config.features.enabled(Feature::RuntimeMetrics), Self::build_model_client_beta_features_header(config.as_ref()), - /*item_ids_enabled*/ config.features.enabled(Feature::ItemIds), + /*item_ids_enabled*/ config.features.enabled(Feature::ItemIds) + || matches!( + session_configuration.history_mode, + ThreadHistoryMode::Paginated + ), /*concurrent_reasoning_summaries_enabled*/ config .features .enabled(Feature::ConcurrentReasoningSummaries), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 9828b71e76..d2555cafef 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -67,6 +67,7 @@ use codex_protocol::permissions::FileSystemSandboxPolicy; use codex_protocol::permissions::FileSystemSpecialPath; use codex_protocol::protocol::NonSteerableTurnKind; use codex_protocol::protocol::SandboxPolicy; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::TurnEnvironmentSelections; use codex_protocol::request_permissions::PermissionGrantScope; use codex_protocol::request_permissions::RequestPermissionProfile; @@ -257,6 +258,24 @@ fn assign_missing_response_item_ids_assigns_additional_tools_ids() { assert!(items[0].id().is_some_and(|id| id.starts_with("at_"))); } +#[tokio::test] +async fn paginated_turn_context_assigns_missing_response_item_ids_without_feature() { + let (session, mut turn_context) = make_session_and_context().await; + turn_context.history_mode = ThreadHistoryMode::Paginated; + let response_item = user_message("hello"); + + let items = session.prepare_conversation_items_for_history( + &turn_context, + std::slice::from_ref(&response_item), + ); + + assert!( + items[0] + .id() + .is_some_and(|item_id| item_id.starts_with("msg_")) + ); +} + fn assistant_message(text: &str) -> ResponseItem { ResponseItem::Message { id: None, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index a7959d1360..608f1c12f5 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -1916,6 +1916,21 @@ async fn drain_in_flight( Ok(()) } +fn assign_missing_streamed_response_item_id( + item: &mut ResponseItem, + active_item: Option<&TurnItem>, +) { + if item.id().is_some() { + return; + } + + let active_item_id = active_item + .map(TurnItem::id) + .filter(|item_id| !item_id.is_empty()); + item.set_id(active_item_id); + Session::assign_missing_response_item_id(item); +} + #[allow(clippy::too_many_arguments)] #[instrument(level = "trace", skip_all, @@ -2030,7 +2045,10 @@ async fn try_run_sampling_request( match event { ResponseEvent::Created => {} - ResponseEvent::OutputItemDone(item) => { + ResponseEvent::OutputItemDone(mut item) => { + if turn_context.item_ids_enabled() { + assign_missing_streamed_response_item_id(&mut item, active_item.as_ref()); + } if let Some((_, mut consumer)) = active_tool_argument_diff_consumer.take() && let Ok(Some(event)) = consumer.finish() { @@ -2124,7 +2142,10 @@ async fn try_run_sampling_request( }); } } - ResponseEvent::OutputItemAdded(item) => { + ResponseEvent::OutputItemAdded(mut item) => { + if turn_context.item_ids_enabled() { + assign_missing_streamed_response_item_id(&mut item, /*active_item*/ None); + } if let ResponseItem::CustomToolCall { call_id, name, diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 1bbffca8c4..e200413d3d 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -10,6 +10,7 @@ use codex_protocol::ThreadId; use codex_protocol::models::AdditionalPermissionProfile; use codex_protocol::openai_models::ModelInfo; use codex_protocol::protocol::MultiAgentVersion; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::TurnEnvironmentSelection; use codex_sandboxing::compatibility_sandbox_policy_for_permission_profile; use codex_sandboxing::policy_transforms::effective_file_system_sandbox_policy; @@ -111,6 +112,7 @@ pub struct TurnContext { pub(crate) reasoning_effort: Option, pub(crate) reasoning_summary: ReasoningSummaryConfig, pub(crate) session_source: SessionSource, + pub(crate) history_mode: ThreadHistoryMode, pub(crate) parent_thread_id: Option, pub(crate) originator: String, pub(crate) environments: TurnEnvironmentSnapshot, @@ -149,6 +151,11 @@ enum TurnMultiAgentRuntime { } impl TurnContext { + pub(crate) fn item_ids_enabled(&self) -> bool { + self.config.features.enabled(Feature::ItemIds) + || matches!(self.history_mode, ThreadHistoryMode::Paginated) + } + pub(crate) fn permission_profile(&self) -> PermissionProfile { self.permission_profile.clone() } @@ -265,6 +272,7 @@ impl TurnContext { reasoning_effort, reasoning_summary: self.reasoning_summary, session_source: self.session_source.clone(), + history_mode: self.history_mode, parent_thread_id: self.parent_thread_id, originator: self.originator.clone(), environments: self.environments.clone(), @@ -544,6 +552,7 @@ impl Session { reasoning_effort, reasoning_summary, session_source, + history_mode: session_configuration.history_mode, parent_thread_id: session_configuration.parent_thread_id, originator: session_configuration.originator.clone(), environments, diff --git a/codex-rs/core/tests/suite/items.rs b/codex-rs/core/tests/suite/items.rs index 196277394a..8b3b4f3008 100644 --- a/codex-rs/core/tests/suite/items.rs +++ b/codex-rs/core/tests/suite/items.rs @@ -276,6 +276,76 @@ async fn reasoning_item_is_emitted() -> anyhow::Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn missing_streamed_reasoning_id_is_reused_for_completion() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let TestCodex { codex, .. } = test_codex() + .with_config(|config| { + let _ = config.features.enable(Feature::ItemIds); + }) + .build_with_auto_env(&server) + .await?; + + let mut reasoning_added = ev_reasoning_item_added("unused", &[]); + reasoning_added["item"] + .as_object_mut() + .expect("reasoning item") + .remove("id"); + let mut reasoning_done = ev_reasoning_item("unused", &["Consider inputs"], &[]); + reasoning_done["item"] + .as_object_mut() + .expect("reasoning item") + .remove("id"); + let response_mock = mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-1"), + reasoning_added, + reasoning_done, + ev_completed("resp-1"), + ]), + ) + .await; + + codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "explain your reasoning".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + + let started_id = wait_for_event_match(&codex, |ev| match ev { + EventMsg::ItemStarted(ItemStartedEvent { + item: TurnItem::Reasoning(item), + .. + }) => Some(item.id.clone()), + _ => None, + }) + .await; + let completed_id = wait_for_event_match(&codex, |ev| match ev { + EventMsg::ItemCompleted(ItemCompletedEvent { + item: TurnItem::Reasoning(item), + .. + }) => Some(item.id.clone()), + _ => None, + }) + .await; + + assert!(started_id.starts_with("rs_")); + assert_eq!(started_id, completed_id); + response_mock.single_request(); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn web_search_item_is_emitted() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/protocol/src/legacy_events.rs b/codex-rs/protocol/src/legacy_events.rs index 0d0dee50f8..63d2c41544 100644 --- a/codex-rs/protocol/src/legacy_events.rs +++ b/codex-rs/protocol/src/legacy_events.rs @@ -75,10 +75,10 @@ impl ContextCompactionItem { } impl UserMessageItem { - pub fn as_legacy_event(&self) -> EventMsg { + pub fn as_legacy_user_message_event(&self) -> UserMessageEvent { // Legacy user-message events flatten only text inputs into `message` and // rebase text element ranges onto that concatenated text. - EventMsg::UserMessage(UserMessageEvent { + UserMessageEvent { client_id: self.client_id.clone(), message: self.message(), images: Some(self.image_urls()), @@ -86,7 +86,11 @@ impl UserMessageItem { local_images: self.local_image_paths(), local_image_details: self.local_image_details(), text_elements: self.text_elements(), - }) + } + } + + pub fn as_legacy_event(&self) -> EventMsg { + EventMsg::UserMessage(self.as_legacy_user_message_event()) } } diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 894c9696bc..57a36c3312 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -120,6 +120,14 @@ pub const CONTEXT_WINDOW_GUIDANCE_OPEN_TAG: &str = ""; pub const CONTEXT_WINDOW_GUIDANCE_CLOSE_TAG: &str = ""; pub const USER_MESSAGE_BEGIN: &str = "## My request for Codex:"; +/// Removes the model-context prefix from a user message before displaying it. +pub fn strip_user_message_prefix(text: &str) -> &str { + match text.find(USER_MESSAGE_BEGIN) { + Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(), + None => text.trim(), + } +} + // TODO(anp): Replace `TurnEnvironmentSelection` with `PathUri` once path URIs carry environment // identifiers. #[derive(Debug, Clone, PartialEq)] @@ -2294,6 +2302,23 @@ pub struct UserMessageEvent { pub text_elements: Vec, } +/// Returns the user-facing preview text for a user message. +pub fn user_message_preview(user: &UserMessageEvent) -> Option { + let message = strip_user_message_prefix(user.message.as_str()); + if !message.is_empty() { + return Some(message.to_string()); + } + if user + .images + .as_ref() + .is_some_and(|images| !images.is_empty()) + || !user.local_images.is_empty() + { + return Some("[Image]".to_string()); + } + None +} + #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] pub struct AgentReasoningEvent { pub text: String, @@ -2591,11 +2616,11 @@ impl InitialHistory { pub fn get_history_mode(&self, default_history_mode: ThreadHistoryMode) -> ThreadHistoryMode { match self { - InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_) => { - default_history_mode - } - InitialHistory::Resumed(_) => self - .get_resumed_session_meta() + InitialHistory::New | InitialHistory::Cleared => default_history_mode, + // Forks copy their source rollout items as-is, so they must keep + // the source format instead of converting to the destination default. + InitialHistory::Resumed(_) | InitialHistory::Forked(_) => self + .get_session_meta() .map(|meta| meta.history_mode) .unwrap_or(default_history_mode), } @@ -5782,13 +5807,13 @@ mod tests { } #[test] - fn resumed_history_uses_persisted_history_mode() -> Result<()> { + fn copied_history_uses_persisted_history_mode() -> Result<()> { let thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000001")?; let session_meta = RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { session_id: thread_id.into(), id: thread_id, - history_mode: ThreadHistoryMode::Paginated, + history_mode: ThreadHistoryMode::Legacy, ..SessionMeta::default() }, git: None, @@ -5800,11 +5825,12 @@ mod tests { }); assert_eq!( - history.get_history_mode(ThreadHistoryMode::Legacy), - ThreadHistoryMode::Paginated + history.get_history_mode(ThreadHistoryMode::Paginated), + ThreadHistoryMode::Legacy ); assert_eq!( - InitialHistory::Forked(vec![session_meta]).get_history_mode(ThreadHistoryMode::Legacy), + InitialHistory::Forked(vec![session_meta]) + .get_history_mode(ThreadHistoryMode::Paginated), ThreadHistoryMode::Legacy ); assert_eq!( diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index 1154d36eee..4729695dcb 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -22,12 +22,13 @@ use crate::protocol::EventMsg; use crate::state_db; use codex_file_search as file_search; use codex_protocol::ThreadId; +use codex_protocol::items::TurnItem; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadHistoryMode; -use codex_protocol::protocol::USER_MESSAGE_BEGIN; +use codex_protocol::protocol::user_message_preview; use serde_json::Value; /// Returned page of thread (thread) summaries. @@ -1184,12 +1185,19 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result { if let Some(preview) = event_msg_preview(&ev) { + // Legacy rollouts persist UserMessage while paginated rollouts persist + // ItemCompleted(UserMessage), so summaries must recognize both formats. + let is_user_message = match &ev { + EventMsg::UserMessage(_) => true, + EventMsg::ItemCompleted(event) => { + matches!(event.item, TurnItem::UserMessage(_)) + } + _ => false, + }; if summary.preview.is_none() { summary.preview = Some(preview.clone()); } - if let EventMsg::UserMessage(_) = ev - && summary.first_user_message.is_none() - { + if is_user_message && summary.first_user_message.is_none() { summary.first_user_message = Some(preview); } } @@ -1250,30 +1258,15 @@ pub async fn read_head_for_summary(path: &Path) -> io::Result &str { - match text.find(USER_MESSAGE_BEGIN) { - Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(), - None => text.trim(), - } -} - fn event_msg_preview(event: &EventMsg) -> Option { match event { - EventMsg::UserMessage(user) => { - let message = strip_user_message_prefix(user.message.as_str()); - if !message.is_empty() { - return Some(message.to_string()); + EventMsg::UserMessage(user) => user_message_preview(user), + EventMsg::ItemCompleted(event) => match &event.item { + TurnItem::UserMessage(user) => { + user_message_preview(&user.as_legacy_user_message_event()) } - if user - .images - .as_ref() - .is_some_and(|images| !images.is_empty()) - || !user.local_images.is_empty() - { - return Some("[Image]".to_string()); - } - None - } + _ => None, + }, EventMsg::ThreadGoalUpdated(event) => { let objective = event.goal.objective.trim(); (!objective.is_empty()).then(|| objective.to_string()) diff --git a/codex-rs/rollout/src/persistence_metrics.rs b/codex-rs/rollout/src/persistence_metrics.rs index 07d16c1334..a3ee1dcb18 100644 --- a/codex-rs/rollout/src/persistence_metrics.rs +++ b/codex-rs/rollout/src/persistence_metrics.rs @@ -8,6 +8,7 @@ use codex_protocol::items::TurnItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::ThreadHistoryMode; use crate::policy::is_persisted_rollout_item; @@ -95,6 +96,7 @@ struct TurnMeasurementUpdate { /// Measures logical JSON sizes while applying the shared rollout persistence policy once. pub fn measure_and_filter_rollout_items( items: &[RolloutItem], + history_mode: ThreadHistoryMode, ) -> (Vec, RolloutPersistenceBatchMeasurement) { let mut persisted = Vec::new(); let mut measurement = RolloutPersistenceBatchMeasurement { @@ -103,7 +105,7 @@ pub fn measure_and_filter_rollout_items( }; for item in items { - let kept = is_persisted_rollout_item(item); + let kept = is_persisted_rollout_item(item, history_mode); let decision = if kept { PersistenceDecision::Kept } else { diff --git a/codex-rs/rollout/src/persistence_metrics_tests.rs b/codex-rs/rollout/src/persistence_metrics_tests.rs index 963805f43d..c5f6d6c9c8 100644 --- a/codex-rs/rollout/src/persistence_metrics_tests.rs +++ b/codex-rs/rollout/src/persistence_metrics_tests.rs @@ -1,11 +1,17 @@ use codex_protocol::ThreadId; +use codex_protocol::items::EnteredReviewModeItem; +use codex_protocol::items::ExitedReviewModeItem; use codex_protocol::items::TurnItem; use codex_protocol::items::UserMessageItem; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::EnteredReviewModeEvent; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::ExitedReviewModeEvent; use codex_protocol::protocol::ItemCompletedEvent; +use codex_protocol::protocol::ReviewTarget; use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; use codex_protocol::protocol::TurnCompleteEvent; @@ -65,7 +71,7 @@ fn update_for_batch( state: &mut TurnMeasurementState, items: &[RolloutItem], ) -> super::TurnMeasurementUpdate { - let (_, measurement) = measure_and_filter_rollout_items(items); + let (_, measurement) = measure_and_filter_rollout_items(items, ThreadHistoryMode::Legacy); update_turn_measurements(state, items, &measurement) } @@ -100,7 +106,8 @@ fn mixed_batch_reports_exact_policy_counts_and_bytes() { let dropped = RolloutItem::ResponseItem(ResponseItem::Other); let items = vec![kept.clone(), dropped.clone()]; - let (persisted, measurement) = measure_and_filter_rollout_items(&items); + let (persisted, measurement) = + measure_and_filter_rollout_items(&items, ThreadHistoryMode::Legacy); let kept_bytes = serde_json::to_vec(&kept) .expect("serialize kept item") .len() as u64; @@ -128,7 +135,8 @@ fn mixed_batch_reports_exact_policy_counts_and_bytes() { #[test] fn retained_items_are_byte_identical() { let item = retained_message("a moderately sized payload"); - let (persisted, measurement) = measure_and_filter_rollout_items(std::slice::from_ref(&item)); + let (persisted, measurement) = + measure_and_filter_rollout_items(std::slice::from_ref(&item), ThreadHistoryMode::Legacy); assert_eq!( serde_json::to_vec(&persisted[0]).expect("serialize persisted item"), @@ -155,8 +163,10 @@ fn turn_measurements_span_batches_and_include_items_before_start() { retained_message("second response"), turn_aborted("turn-2"), ]; - let (_, first_expected) = measure_and_filter_rollout_items(&first_turn); - let (_, second_expected) = measure_and_filter_rollout_items(&second_turn); + let (_, first_expected) = + measure_and_filter_rollout_items(&first_turn, ThreadHistoryMode::Legacy); + let (_, second_expected) = + measure_and_filter_rollout_items(&second_turn, ThreadHistoryMode::Legacy); let batches = [ first_turn[..1].to_vec(), first_turn[1..3].to_vec(), @@ -217,7 +227,8 @@ fn invalid_turn_boundaries_reset_partial_measurements() { retained_message("retained turn"), turn_complete("turn-2"), ]; - let (_, expected) = measure_and_filter_rollout_items(&replacement[2..]); + let (_, expected) = + measure_and_filter_rollout_items(&replacement[2..], ThreadHistoryMode::Legacy); let update = update_for_batch(&mut state, &replacement); assert_eq!( @@ -235,7 +246,7 @@ fn invalid_turn_boundaries_reset_partial_measurements() { } #[test] -fn filtered_item_completion_includes_its_nested_item_type() { +fn item_completion_persistence_depends_on_history_mode() { let item = RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent { thread_id: ThreadId::default(), turn_id: "turn".to_string(), @@ -247,14 +258,87 @@ fn filtered_item_completion_includes_its_nested_item_type() { completed_at_ms: 0, })); - let (_, measurement) = measure_and_filter_rollout_items(&[item]); + let (_, legacy_measurement) = + measure_and_filter_rollout_items(std::slice::from_ref(&item), ThreadHistoryMode::Legacy); assert_eq!( - measurement.items[0].rollout_item_type, + legacy_measurement.items[0].rollout_item_type, "event.item_completed.user_message" ); assert_eq!( - measurement.items[0].decision, + legacy_measurement.items[0].decision, super::PersistenceDecision::Dropped ); + + let (persisted, paginated_measurement) = + measure_and_filter_rollout_items(std::slice::from_ref(&item), ThreadHistoryMode::Paginated); + + assert_eq!( + serde_json::to_value(persisted).expect("serialize persisted items"), + serde_json::to_value([item]).expect("serialize expected items") + ); + assert_eq!( + paginated_measurement.items[0].decision, + super::PersistenceDecision::Kept + ); +} + +#[test] +fn review_mode_persistence_depends_on_history_mode() { + let completed_items = vec![ + RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent { + thread_id: ThreadId::default(), + turn_id: "turn".to_string(), + item: TurnItem::EnteredReviewMode(EnteredReviewModeItem { + id: "entered-review".to_string(), + target: ReviewTarget::Custom { + instructions: "review this".to_string(), + }, + user_facing_hint: "Review requested.".to_string(), + }), + completed_at_ms: 0, + })), + RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent { + thread_id: ThreadId::default(), + turn_id: "turn".to_string(), + item: TurnItem::ExitedReviewMode(ExitedReviewModeItem { + id: "exited-review".to_string(), + review_output: None, + }), + completed_at_ms: 0, + })), + ]; + let legacy_events = vec![ + RolloutItem::EventMsg(EventMsg::EnteredReviewMode(EnteredReviewModeEvent { + target: ReviewTarget::Custom { + instructions: "review this".to_string(), + }, + user_facing_hint: Some("Review requested.".to_string()), + turn_id: Some("turn".to_string()), + item_id: Some("entered-review".to_string()), + })), + RolloutItem::EventMsg(EventMsg::ExitedReviewMode(ExitedReviewModeEvent { + turn_id: Some("turn".to_string()), + item_id: Some("exited-review".to_string()), + review_output: None, + })), + ]; + let items = completed_items + .iter() + .chain(&legacy_events) + .cloned() + .collect::>(); + + let (persisted_legacy, _) = measure_and_filter_rollout_items(&items, ThreadHistoryMode::Legacy); + assert_eq!( + serde_json::to_value(persisted_legacy).expect("serialize persisted items"), + serde_json::to_value(legacy_events).expect("serialize expected items") + ); + + let (persisted_paginated, _) = + measure_and_filter_rollout_items(&items, ThreadHistoryMode::Paginated); + assert_eq!( + serde_json::to_value(persisted_paginated).expect("serialize persisted items"), + serde_json::to_value(completed_items).expect("serialize expected items") + ); } diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 33103f54c9..75ddec35ae 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -1,14 +1,16 @@ use crate::protocol::EventMsg; use crate::protocol::RolloutItem; +use codex_protocol::items::TurnItem; use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::ThreadHistoryMode; /// Whether a rollout `item` should be persisted in rollout files. -pub fn is_persisted_rollout_item(item: &RolloutItem) -> bool { +pub fn is_persisted_rollout_item(item: &RolloutItem, history_mode: ThreadHistoryMode) -> bool { match item { RolloutItem::ResponseItem(item) => should_persist_response_item(item), RolloutItem::InterAgentCommunication(_) | RolloutItem::InterAgentCommunicationMetadata { .. } => true, - RolloutItem::EventMsg(ev) => should_persist_event_msg(ev), + RolloutItem::EventMsg(ev) => should_persist_event_msg(ev, history_mode), // Persist Codex executive markers so we can analyze flows (e.g., compaction, API turns). RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) @@ -17,11 +19,14 @@ pub fn is_persisted_rollout_item(item: &RolloutItem) -> bool { } } -/// Return the canonical rollout items that should be persisted for a live append. -pub fn persisted_rollout_items(items: &[RolloutItem]) -> Vec { +/// Return the rollout items that should be persisted for a live append. +pub fn persisted_rollout_items( + items: &[RolloutItem], + history_mode: ThreadHistoryMode, +) -> Vec { let mut persisted = Vec::new(); for item in items { - if is_persisted_rollout_item(item) { + if is_persisted_rollout_item(item, history_mode) { persisted.push(item.clone()); } } @@ -78,37 +83,38 @@ pub fn should_persist_response_item_for_memories(item: &ResponseItem) -> bool { /// Whether an `EventMsg` should be persisted in rollout files. #[inline] -pub fn should_persist_event_msg(ev: &EventMsg) -> bool { +pub fn should_persist_event_msg(ev: &EventMsg, history_mode: ThreadHistoryMode) -> bool { match ev { - EventMsg::UserMessage(_) - | EventMsg::AgentMessage(_) - | EventMsg::AgentReasoning(_) - | EventMsg::AgentReasoningRawContent(_) - | EventMsg::PatchApplyEnd(_) - | EventMsg::TokenCount(_) + EventMsg::ItemCompleted(event) => { + // Paginated rollouts store TurnItems. + // Legacy rollouts keep only items with no raw ResponseItem or legacy equivalent. + matches!(history_mode, ThreadHistoryMode::Paginated) + || matches!(event.item, TurnItem::Plan(_) | TurnItem::Sleep(_)) + } + EventMsg::TokenCount(_) | EventMsg::ThreadGoalUpdated(_) - | EventMsg::ContextCompacted(_) - | EventMsg::EnteredReviewMode(_) - | EventMsg::ExitedReviewMode(_) - | EventMsg::McpToolCallEnd(_) | EventMsg::ThreadRolledBack(_) | EventMsg::TurnAborted(_) | EventMsg::TurnStarted(_) | EventMsg::TurnComplete(_) - | EventMsg::ThreadSettingsApplied(_) + | EventMsg::ThreadSettingsApplied(_) => true, + + // Only persist these legacy events when the thread's history mode is Legacy. + // New, paginated rollouts persist ItemCompleted events with TurnItems. + EventMsg::UserMessage(_) + | EventMsg::AgentMessage(_) + | EventMsg::AgentReasoning(_) + | EventMsg::AgentReasoningRawContent(_) + | EventMsg::EnteredReviewMode(_) + | EventMsg::ExitedReviewMode(_) + | EventMsg::PatchApplyEnd(_) + | EventMsg::ContextCompacted(_) + | EventMsg::McpToolCallEnd(_) | EventMsg::WebSearchEnd(_) | EventMsg::ImageGenerationEnd(_) - | EventMsg::SubAgentActivity(_) => true, - EventMsg::ItemCompleted(event) => { - // These items have no equivalent raw ResponseItem or legacy event, - // so persist their completion for replay without retaining every - // item lifecycle event. - matches!( - event.item, - codex_protocol::items::TurnItem::Plan(_) - | codex_protocol::items::TurnItem::Sleep(_) - ) - } + | EventMsg::SubAgentActivity(_) => matches!(history_mode, ThreadHistoryMode::Legacy), + + // Transient, non-durable events. EventMsg::Error(_) | EventMsg::GuardianAssessment(_) | EventMsg::ExecCommandEnd(_) diff --git a/codex-rs/rollout/src/search.rs b/codex-rs/rollout/src/search.rs index 58b46887a9..805a6b274e 100644 --- a/codex-rs/rollout/src/search.rs +++ b/codex-rs/rollout/src/search.rs @@ -9,7 +9,7 @@ use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; -use codex_protocol::protocol::USER_MESSAGE_BEGIN; +use codex_protocol::protocol::strip_user_message_prefix; use regex::Regex; use regex::RegexBuilder; use tokio::process::Command; @@ -297,13 +297,6 @@ fn content_item_text(item: &ContentItem) -> Option<&str> { } } -fn strip_user_message_prefix(text: &str) -> &str { - match text.find(USER_MESSAGE_BEGIN) { - Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(), - None => text.trim(), - } -} - fn excerpt_around_match(text: &str, search_term: &Regex) -> Option { let normalized = normalize_preview_text(text); let matched = search_term.find(normalized.as_str())?; diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index e98ef0d763..ac4201de9f 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -1,16 +1,16 @@ use crate::model::ThreadMetadata; +use codex_protocol::items::TurnItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::TurnContextItem; -use codex_protocol::protocol::USER_MESSAGE_BEGIN; use codex_protocol::protocol::UserMessageEvent; +use codex_protocol::protocol::strip_user_message_prefix; +use codex_protocol::protocol::user_message_preview; use serde::Serialize; use serde_json::Value; -const IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER: &str = "[Image]"; - /// Apply a rollout item to the metadata structure. pub fn apply_rollout_item( metadata: &mut ThreadMetadata, @@ -39,6 +39,11 @@ pub fn rollout_item_affects_thread_metadata(item: &RolloutItem) -> bool { RolloutItem::EventMsg( EventMsg::TokenCount(_) | EventMsg::UserMessage(_) | EventMsg::ThreadGoalUpdated(_), ) => true, + RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) + if matches!(event.item, TurnItem::UserMessage(_)) => + { + true + } RolloutItem::EventMsg(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) @@ -96,16 +101,11 @@ fn apply_event_msg(metadata: &mut ThreadMetadata, event: &EventMsg) { } } EventMsg::UserMessage(user) => { - let preview = user_message_preview(user); - if metadata.first_user_message.is_none() { - metadata.first_user_message = preview.clone(); - } - set_preview_if_empty(metadata, preview); - if metadata.title.is_empty() { - let title = strip_user_message_prefix(user.message.as_str()); - if !title.is_empty() { - metadata.title = title.to_string(); - } + apply_user_message(metadata, user); + } + EventMsg::ItemCompleted(event) => { + if let TurnItem::UserMessage(user) = &event.item { + apply_user_message(metadata, &user.as_legacy_user_message_event()); } } EventMsg::ThreadGoalUpdated(event) => { @@ -120,35 +120,26 @@ fn apply_event_msg(metadata: &mut ThreadMetadata, event: &EventMsg) { fn apply_response_item(_metadata: &mut ThreadMetadata, _item: &ResponseItem) {} +fn apply_user_message(metadata: &mut ThreadMetadata, user: &UserMessageEvent) { + let preview = user_message_preview(user); + if metadata.first_user_message.is_none() { + metadata.first_user_message = preview.clone(); + } + set_preview_if_empty(metadata, preview); + if metadata.title.is_empty() { + let title = strip_user_message_prefix(user.message.as_str()); + if !title.is_empty() { + metadata.title = title.to_string(); + } + } +} + fn set_preview_if_empty(metadata: &mut ThreadMetadata, preview: Option) { if metadata.preview.is_none() { metadata.preview = preview; } } -fn strip_user_message_prefix(text: &str) -> &str { - match text.find(USER_MESSAGE_BEGIN) { - Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(), - None => text.trim(), - } -} - -fn user_message_preview(user: &UserMessageEvent) -> Option { - let message = strip_user_message_prefix(user.message.as_str()); - if !message.is_empty() { - return Some(message.to_string()); - } - if user - .images - .as_ref() - .is_some_and(|images| !images.is_empty()) - || !user.local_images.is_empty() - { - return Some(IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER.to_string()); - } - None -} - pub(crate) fn enum_to_string(value: &T) -> String { match serde_json::to_value(value) { Ok(Value::String(s)) => s, @@ -164,12 +155,15 @@ mod tests { use chrono::DateTime; use chrono::Utc; use codex_protocol::ThreadId; + use codex_protocol::items::TurnItem; + use codex_protocol::items::UserMessageItem; use codex_protocol::models::ContentItem; use codex_protocol::models::PermissionProfile; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::EventMsg; + use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionMeta; @@ -182,6 +176,7 @@ mod tests { use codex_protocol::protocol::TurnContextItem; use codex_protocol::protocol::USER_MESSAGE_BEGIN; use codex_protocol::protocol::UserMessageEvent; + use codex_protocol::user_input::UserInput; use pretty_assertions::assert_eq; use std::path::PathBuf; @@ -229,6 +224,29 @@ mod tests { assert_eq!(metadata.title, "actual user request"); } + #[test] + fn completed_user_message_items_set_title_and_first_user_message() { + let mut metadata = metadata_for_test(); + let item = RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent { + thread_id: ThreadId::default(), + turn_id: "turn-1".to_string(), + item: TurnItem::UserMessage(UserMessageItem::new(&[UserInput::Text { + text: format!("{USER_MESSAGE_BEGIN} actual user request"), + text_elements: Vec::new(), + }])), + completed_at_ms: 0, + })); + + apply_rollout_item(&mut metadata, &item, "test-provider"); + + assert_eq!( + metadata.first_user_message.as_deref(), + Some("actual user request") + ); + assert_eq!(metadata.preview.as_deref(), Some("actual user request")); + assert_eq!(metadata.title, "actual user request"); + } + #[test] fn event_msg_image_only_user_message_sets_image_placeholder_preview() { let mut metadata = metadata_for_test(); @@ -243,14 +261,8 @@ mod tests { apply_rollout_item(&mut metadata, &item, "test-provider"); - assert_eq!( - metadata.first_user_message.as_deref(), - Some(super::IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER) - ); - assert_eq!( - metadata.preview.as_deref(), - Some(super::IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER) - ); + assert_eq!(metadata.first_user_message.as_deref(), Some("[Image]")); + assert_eq!(metadata.preview.as_deref(), Some("[Image]")); assert_eq!(metadata.title, ""); } diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index cb3f198ff3..0d6ea80f2a 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -482,17 +482,21 @@ impl InMemoryThreadStore { } async fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreResult<()> { - let canonical_items = persisted_rollout_items(params.items.as_slice()); - if canonical_items.is_empty() { + if params.items.is_empty() { return Ok(()); } let mut state = self.state.lock().await; + let history_mode = history_mode_from_state(&state, params.thread_id); + let persisted_items = persisted_rollout_items(params.items.as_slice(), history_mode); + if persisted_items.is_empty() { + return Ok(()); + } state.calls.append_items += 1; state .histories .entry(params.thread_id) .or_default() - .extend(canonical_items); + .extend(persisted_items); Ok(()) } diff --git a/codex-rs/thread-store/src/live_thread.rs b/codex-rs/thread-store/src/live_thread.rs index 793bbf9965..fdc481642b 100644 --- a/codex-rs/thread-store/src/live_thread.rs +++ b/codex-rs/thread-store/src/live_thread.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use codex_protocol::ThreadId; use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadMemoryMode; use codex_rollout::RolloutPersistenceTelemetry; use codex_rollout::measure_and_filter_rollout_items; @@ -32,6 +33,7 @@ use crate::thread_metadata_sync::ThreadMetadataSync; #[derive(Clone)] pub struct LiveThread { thread_id: ThreadId, + history_mode: ThreadHistoryMode, thread_store: Arc, metadata_sync: Arc>, persistence_telemetry: RolloutPersistenceTelemetry, @@ -92,10 +94,12 @@ impl LiveThread { params: CreateThreadParams, ) -> ThreadStoreResult { let thread_id = params.thread_id; + let history_mode = params.history_mode; let metadata_sync = ThreadMetadataSync::for_create(¶ms).await; thread_store.create_thread(params).await?; Ok(Self { thread_id, + history_mode, thread_store, metadata_sync: Arc::new(Mutex::new(metadata_sync)), persistence_telemetry: RolloutPersistenceTelemetry::new(thread_id), @@ -104,6 +108,7 @@ impl LiveThread { pub async fn resume( thread_store: Arc, + history_mode: ThreadHistoryMode, params: ResumeThreadParams, ) -> ThreadStoreResult { let thread_id = params.thread_id; @@ -132,6 +137,7 @@ impl LiveThread { } Ok(Self { thread_id, + history_mode, thread_store, metadata_sync: Arc::new(Mutex::new(metadata_sync)), persistence_telemetry: RolloutPersistenceTelemetry::new(thread_id), @@ -141,36 +147,38 @@ impl LiveThread { #[tracing::instrument( level = "trace", skip_all, - fields(item_count = items.len()) + fields(item_count = raw_items.len()) )] - pub async fn append_items(&self, items: &[RolloutItem]) -> ThreadStoreResult<()> { + pub async fn append_items(&self, raw_items: &[RolloutItem]) -> ThreadStoreResult<()> { // Empty appends are intentionally ignored rather than represented as zero-sized batches. - if items.is_empty() { + if raw_items.is_empty() { return Ok(()); } - let (canonical_items, measurement) = if self.persistence_telemetry.is_enabled() { - let (canonical_items, measurement) = measure_and_filter_rollout_items(items); - (canonical_items, Some(measurement)) + let (items, measurement) = if self.persistence_telemetry.is_enabled() { + let (items, measurement) = + measure_and_filter_rollout_items(raw_items, self.history_mode); + (items, Some(measurement)) } else { - (persisted_rollout_items(items), None) + (persisted_rollout_items(raw_items, self.history_mode), None) }; self.thread_store .append_items(AppendThreadItemsParams { thread_id: self.thread_id, - items: items.to_vec(), + items: raw_items.to_vec(), }) .await?; if let Some(measurement) = measurement.as_ref() { - self.persistence_telemetry.record_batch(items, measurement); + self.persistence_telemetry + .record_batch(raw_items, measurement); } - if canonical_items.is_empty() { + if items.is_empty() { return Ok(()); } let update = self .metadata_sync .lock() .await - .observe_appended_items(canonical_items.as_slice()); + .observe_appended_items(items.as_slice()); if let Some(update) = update { self.thread_store .update_thread_metadata(UpdateThreadMetadataParams { diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index 74e707bb58..3e06bdcb76 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -1,6 +1,7 @@ use std::path::PathBuf; use codex_protocol::ThreadId; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadMemoryMode; use codex_rollout::RolloutConfig; use codex_rollout::RolloutRecorder; @@ -115,13 +116,15 @@ pub(super) async fn append_items( store: &LocalThreadStore, params: AppendThreadItemsParams, ) -> ThreadStoreResult<()> { - let canonical_items = persisted_rollout_items(params.items.as_slice()); - if canonical_items.is_empty() { + // LocalThreadStore rejects paginated threads before opening a writer. + let persisted_items = + persisted_rollout_items(params.items.as_slice(), ThreadHistoryMode::Legacy); + if persisted_items.is_empty() { return Ok(()); } let recorder = store.live_recorder(params.thread_id).await?; recorder - .record_canonical_items(canonical_items.as_slice()) + .record_canonical_items(persisted_items.as_slice()) .await .map_err(thread_store_io_error)?; // LiveThread applies metadata immediately after append_items returns. Wait for the local diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 5939ea606e..627b001c49 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -694,6 +694,7 @@ mod tests { write_session_file(home.path(), "2025-01-03T17-00-00", uuid).expect("session file"); let live_thread = LiveThread::resume( store, + ThreadHistoryMode::Legacy, ResumeThreadParams { thread_id, rollout_path: Some(rollout_path), @@ -748,6 +749,7 @@ mod tests { .expect("external session file"); let live_thread = LiveThread::resume( store, + ThreadHistoryMode::Legacy, ResumeThreadParams { thread_id, rollout_path: Some(rollout_path), diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index 1057715a4a..4b53990828 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -1,5 +1,3 @@ -#[cfg(test)] -use std::sync::Arc; use std::time::Duration; use std::time::Instant; @@ -9,29 +7,27 @@ use chrono::Utc; use codex_git_utils::collect_git_info; use codex_git_utils::get_git_repo_root; use codex_protocol::ThreadId; +use codex_protocol::items::TurnItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::GitInfo; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::ThreadMemoryMode; -use codex_protocol::protocol::USER_MESSAGE_BEGIN; use codex_protocol::protocol::UserMessageEvent; +use codex_protocol::protocol::strip_user_message_prefix; +use codex_protocol::protocol::user_message_preview; use crate::CreateThreadParams; use crate::GitInfoPatch; use crate::ResumeThreadParams; use crate::ThreadMetadataPatch; -const IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER: &str = "[Image]"; -#[cfg(not(test))] const THREAD_UPDATED_AT_TOUCH_INTERVAL: Duration = Duration::from_secs(5); -#[cfg(test)] -const THREAD_UPDATED_AT_TOUCH_INTERVAL: Duration = Duration::from_millis(50); -/// Live-thread helper that derives metadata updates from canonical rollout items. +/// Live-thread helper that derives metadata updates from appended rollout items. /// -/// Stores receive raw history plus explicit metadata patches. This helper keeps append-derived -/// metadata observation in the live layer without owning persistence-policy filtering or making -/// `append_items` infer metadata inside a `ThreadStore` implementation. +/// Stores receive raw rollout items plus explicit metadata patches. This helper +/// keeps append-derived metadata observation in the live layer without owning persistence-policy +/// filtering or making `append_items` infer metadata inside a `ThreadStore` implementation. pub(crate) struct ThreadMetadataSync { thread_id: ThreadId, cwd_seen: bool, @@ -249,22 +245,14 @@ impl ThreadMetadataSync { update.permission_profile = Some(turn_ctx.permission_profile()); } RolloutItem::EventMsg(EventMsg::UserMessage(user)) => { - if let Some(preview) = user_message_preview(user) { - if !self.first_user_message_seen { - self.first_user_message_seen = true; - update.first_user_message = Some(preview.clone()); - } - if !self.preview_seen { - self.preview_seen = true; - update.preview = Some(preview); - } - } - if !self.title_seen { - let title = strip_user_message_prefix(user.message.as_str()); - if !title.is_empty() { - self.title_seen = true; - update.title = Some(title.to_string()); - } + self.observe_user_message(user, &mut update); + } + RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) => { + if let TurnItem::UserMessage(user) = &event.item { + self.observe_user_message( + &user.as_legacy_user_message_event(), + &mut update, + ); } } RolloutItem::EventMsg(EventMsg::TokenCount(token_count)) => { @@ -293,6 +281,26 @@ impl ThreadMetadataSync { Some(update) } + fn observe_user_message(&mut self, user: &UserMessageEvent, update: &mut ThreadMetadataPatch) { + if let Some(preview) = user_message_preview(user) { + if !self.first_user_message_seen { + self.first_user_message_seen = true; + update.first_user_message = Some(preview.clone()); + } + if !self.preview_seen { + self.preview_seen = true; + update.preview = Some(preview); + } + } + if !self.title_seen { + let title = strip_user_message_prefix(user.message.as_str()); + if !title.is_empty() { + self.title_seen = true; + update.title = Some(title.to_string()); + } + } + } + fn merge_pending_update(&mut self, update: Option) { let Some(update) = update else { return; @@ -323,29 +331,6 @@ fn parse_session_timestamp(value: &str) -> Option> { .ok() } -fn strip_user_message_prefix(text: &str) -> &str { - match text.find(USER_MESSAGE_BEGIN) { - Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(), - None => text.trim(), - } -} - -fn user_message_preview(user: &UserMessageEvent) -> Option { - let message = strip_user_message_prefix(user.message.as_str()); - if !message.is_empty() { - return Some(message.to_string()); - } - if user - .images - .as_ref() - .is_some_and(|images| !images.is_empty()) - || !user.local_images.is_empty() - { - return Some(IMAGE_ONLY_USER_MESSAGE_PLACEHOLDER.to_string()); - } - None -} - fn thread_updated_at_touch() -> ThreadMetadataPatch { ThreadMetadataPatch { updated_at: Some(Utc::now()), @@ -387,7 +372,11 @@ fn git_info_patch_from_observation(git_info: GitInfo) -> GitInfoPatch { #[cfg(test)] mod tests { + use std::sync::Arc; + + use codex_protocol::items::UserMessageItem; use codex_protocol::protocol::CompactedItem; + use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; @@ -396,6 +385,7 @@ mod tests { use codex_protocol::protocol::ThreadGoalUpdatedEvent; use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::UserMessageEvent; + use codex_protocol::user_input::UserInput; use pretty_assertions::assert_eq; use super::*; @@ -484,6 +474,32 @@ mod tests { assert!(update.patch.updated_at.is_some()); } + #[test] + fn completed_user_message_items_emit_metadata_fields() { + let thread_id = ThreadId::new(); + let mut sync = ThreadMetadataSync::for_resume(&resume_params(thread_id, Vec::new())); + let update = sync + .observe_appended_items(&[RolloutItem::EventMsg(EventMsg::ItemCompleted( + ItemCompletedEvent { + thread_id, + turn_id: "turn-1".to_string(), + item: TurnItem::UserMessage(UserMessageItem::new(&[UserInput::Text { + text: "first user text".to_string(), + text_elements: Vec::new(), + }])), + completed_at_ms: 0, + }, + ))]) + .expect("completed user message metadata update"); + + assert_eq!(update.patch.preview.as_deref(), Some("first user text")); + assert_eq!(update.patch.title.as_deref(), Some("first user text")); + assert_eq!( + update.patch.first_user_message.as_deref(), + Some("first user text") + ); + } + #[test] fn metadata_irrelevant_items_coalesce_updated_at_touches() { let thread_id = ThreadId::new();