From 2342b2c2a66b5ac2a07278afb4f5f38700be452e Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Wed, 8 Jul 2026 19:55:03 -0700 Subject: [PATCH] feat(rollout): persist TurnItems for paginated thread rollouts (#30188) ## Description This PR makes new threads with `history_mode = "paginated"` persist `ItemCompleted(item: )` in their rollout JSONL file. Legacy threads keep persisting the existing legacy events. Because the format is selected per thread, a rollout is either legacy or paginated; we do not need to support mixed rollouts containing both representations. This PR depends on [#31473](https://github.com/openai/codex/pull/31473). ## Why Paginated thread history needs stable turn/item IDs and completed item snapshots so the later SQLite projector can materialize appended rollout JSONL without rebuilding the whole thread. Keeping the legacy persistence policy unchanged avoids changing historical rollouts or the readers that still consume them. ## What changed - Made rollout filtering history-mode aware. Paginated threads keep completed canonical `ItemCompleted` events and drop their redundant legacy projections; legacy threads keep the existing event set. - Made forks inherit the source thread history mode, so copied legacy history is never filtered as paginated. - Made paginated threads assign IDs to locally-created response items even when `Feature::ItemIds` is off, and reject streamed output items that arrive without server IDs. - Updated legacy turn replay, rollout list/search, and SQLite metadata extraction to understand completed canonical user-message items. --- codex-rs/app-server/src/request_processors.rs | 6 +- .../external_agent_session_import.rs | 2 +- .../request_processors/thread_lifecycle.rs | 2 +- .../request_processors/thread_processor.rs | 11 +- .../src/request_processors/thread_summary.rs | 5 +- .../thread_summary_tests.rs | 1 + codex-rs/core/src/session/mod.rs | 8 +- codex-rs/core/src/session/review.rs | 1 + codex-rs/core/src/session/session.rs | 13 +- codex-rs/core/src/session/tests.rs | 19 +++ codex-rs/core/src/session/turn.rs | 25 +++- codex-rs/core/src/session/turn_context.rs | 9 ++ codex-rs/core/tests/suite/items.rs | 70 +++++++++++ codex-rs/protocol/src/legacy_events.rs | 10 +- codex-rs/protocol/src/protocol.rs | 46 +++++-- codex-rs/rollout/src/list.rs | 43 +++---- codex-rs/rollout/src/persistence_metrics.rs | 4 +- .../rollout/src/persistence_metrics_tests.rs | 104 ++++++++++++++-- codex-rs/rollout/src/policy.rs | 62 +++++----- codex-rs/rollout/src/search.rs | 9 +- codex-rs/state/src/extract.rs | 100 ++++++++------- codex-rs/thread-store/src/in_memory.rs | 10 +- codex-rs/thread-store/src/live_thread.rs | 30 +++-- .../thread-store/src/local/live_writer.rs | 9 +- codex-rs/thread-store/src/local/mod.rs | 2 + .../thread-store/src/thread_metadata_sync.rs | 116 ++++++++++-------- 26 files changed, 497 insertions(+), 220 deletions(-) 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();