diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 2b81d002c4..82dbae7584 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -465,6 +465,7 @@ use codex_protocol::protocol::SessionConfiguredEvent; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::TurnEnvironmentRequest; use codex_protocol::protocol::W3cTraceContext; +#[cfg(test)] 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; 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 734563f903..b4d27cde9a 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -5702,12 +5702,19 @@ impl ThreadRequestProcessor { } } - let ephemeral_preview = if ephemeral { - if paginated_source && last_turn_id.is_none() && before_turn_id.is_none() { - source_thread.preview.clone() - } else { - preview_from_rollout_items(&history_items) - } + let inherited_preview = if paginated_source + && last_turn_id.is_none() + && before_turn_id.is_none() + && !source_thread.preview.is_empty() + { + source_thread.preview.clone() + } else if let Some(preview) = prepared_fork + .as_mut() + .and_then(|fork| fork.preview_before_compaction.take()) + { + preview + } else if paginated_source || ephemeral { + preview_from_rollout_items(&history_items) } else { String::new() }; @@ -5727,6 +5734,7 @@ impl ThreadRequestProcessor { self.thread_manager.as_ref(), self.thread_store.as_ref(), StoreThreadMetadataPatch { + preview: (!inherited_preview.is_empty()).then(|| inherited_preview.clone()), project_id: inherited_project_id.clone().map(Some), daybreak_enabled: source_thread.daybreak_enabled, ..Default::default() @@ -5891,7 +5899,7 @@ impl ThreadRequestProcessor { &config_snapshot, /*path*/ None, ); - thread.preview = ephemeral_preview; + thread.preview = inherited_preview; thread.forked_from_id = Some(source_thread_id.to_string()); thread.turns = ephemeral_turns; (thread, ephemeral_token_usage_turn_id) @@ -6912,12 +6920,15 @@ fn preview_from_rollout_items(items: &[RolloutItem]) -> String { .iter() .find_map(|item| match item { RolloutItem::ResponseItem(item) => match codex_core::parse_turn_item(&item.item) { - Some(codex_protocol::items::TurnItem::UserMessage(user)) => Some(user.message()), + Some(codex_protocol::items::TurnItem::UserMessage(user)) => { + codex_protocol::protocol::user_message_preview( + &user.as_legacy_user_message_event(), + ) + } _ => None, }, _ => None, }) - .map(|preview| strip_user_message_prefix(preview.as_str()).to_string()) .unwrap_or_default() } diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index cb329b7a12..e732b9296f 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -915,7 +915,7 @@ async fn assert_thread_fork_at_named_boundary_keeps_only_terminal_prefix( .with_codex_home(codex_home.path()) .build_initialized() .await?; - let _: ThreadResumeResponse = resumed + let reopened: ThreadResumeResponse = resumed .request(|request_id| ClientRequest::ThreadResume { request_id, params: ThreadResumeParams { @@ -936,6 +936,34 @@ async fn assert_thread_fork_at_named_boundary_keeps_only_terminal_prefix( .await?; assert!(after_resume.data.is_empty()); + if history_mode == ThreadHistoryMode::Paginated { + assert_eq!(reopened.thread.preview, "first"); + // The source is closed. Place the next fork boundary immediately after a compaction + // whose replacement history no longer contains the opening user message. + for item in [ + RolloutItem::Compacted(serde_json::from_value(json!({ + "message": "summary", "replacement_history": [], "window_number": 1 + }))?), + RolloutItem::EventMsg(EventMsg::TurnStarted(serde_json::from_value(json!({ + "turn_id": "after-compaction" + }))?)), + ] { + append_rollout_item_to_path(&source_path, &item).await?; + } + let fork: ThreadForkResponse = resumed + .request(|request_id| ClientRequest::ThreadFork { + request_id, + params: ThreadForkParams { + thread_id: source_thread.id, + before_turn_id: Some("after-compaction".to_string()), + exclude_turns: true, + ..Default::default() + }, + }) + .await?; + assert_eq!(fork.thread.preview, "first"); + } + Ok(()) } @@ -1618,6 +1646,15 @@ async fn thread_fork_creates_reference_backed_paginated_thread() -> Result<()> { assert_eq!(forked_thread.forked_from_id, Some(conversation_id.clone())); assert_eq!(forked_thread.turns.len(), 1); let forked_thread_id = forked_thread.id.clone(); + let listed = list_threads(&mut mcp).await?; + assert_eq!( + listed + .data + .iter() + .find(|thread| thread.id == forked_thread_id) + .map(|thread| thread.preview.as_str()), + Some("Saved user message") + ); let forked_path = forked_thread.path.expect("forked rollout path"); assert!(!std::fs::read_to_string(forked_path.as_path())?.contains("Saved user message")); let meta = read_session_meta_line(forked_path.as_path()).await?; @@ -1643,6 +1680,15 @@ async fn thread_fork_creates_reference_backed_paginated_thread() -> Result<()> { mcp.read_stream_until_notification_message("turn/completed"), ) .await??; + let listed = list_threads(&mut mcp).await?; + assert_eq!( + listed + .data + .iter() + .find(|thread| thread.id == forked_thread_id) + .map(|thread| thread.preview.as_str()), + Some("Saved user message") + ); let requests = server.received_requests().await.expect("wiremock requests"); let response_request = requests .iter() diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index 38f5174ffb..ec40f50ea6 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -122,6 +122,7 @@ pub type ConversationsPage = ThreadsPage; struct HeadTailSummary { originator: Option, saw_session_meta: bool, + has_fork_history: bool, attached_to_parent: bool, thread_id: Option, first_user_message: Option, @@ -841,7 +842,9 @@ async fn build_thread_item( return None; } if summary.saw_session_meta - && (matches!(preview_filter, PreviewFilter::IncludeEmpty) || summary.preview.is_some()) + && (matches!(preview_filter, PreviewFilter::IncludeEmpty) + || summary.preview.is_some() + || summary.has_fork_history) { let HeadTailSummary { originator, @@ -1033,6 +1036,13 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result 0) + && !session_meta_line.meta.source.is_non_root_agent(); summary.history_mode = session_meta_line.meta.history_mode; summary.parent_thread_id = session_meta_line.meta.parent_thread_id; summary.agent_nickname = session_meta_line.meta.agent_nickname.clone(); diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 8faee09898..c590d52932 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -1431,12 +1431,20 @@ fn fill_missing_thread_item_metadata(item: &mut ThreadItem, state_item: ThreadIt item.originator = originator; } + // A paginated fork's rollout starts with its own first turn. SQLite can hold a + // distinct preview inherited from the parent before that turn was written. + if item.preview.is_none() + || (item.history_mode == ThreadHistoryMode::Paginated + && preview + .as_deref() + .is_some_and(|preview| !preview.is_empty()) + && preview != first_user_message) + { + item.preview = preview; + } if item.first_user_message.is_none() { item.first_user_message = first_user_message; } - if item.preview.is_none() { - item.preview = preview; - } item.section = section; item.project_id = project_id; item.daybreak_enabled = daybreak_enabled; diff --git a/codex-rs/thread-store/src/live_thread.rs b/codex-rs/thread-store/src/live_thread.rs index 0c45c2ea6d..a104a15ab5 100644 --- a/codex-rs/thread-store/src/live_thread.rs +++ b/codex-rs/thread-store/src/live_thread.rs @@ -138,7 +138,16 @@ impl LiveThread { ) -> ThreadStoreResult { let thread_id = params.thread_id; let history_mode = params.history_mode; - let metadata_sync = ThreadMetadataSync::for_create(¶ms).await; + let inherited_preview = if params.forked_from_id.is_some() { + thread_store + .read_pending_thread_metadata(thread_id) + .await? + .and_then(|patch| patch.preview) + } else { + None + }; + let metadata_sync = + ThreadMetadataSync::for_create(¶ms, inherited_preview.as_deref()).await; thread_store.create_thread(params).await?; Ok(Self { thread_id, diff --git a/codex-rs/thread-store/src/local/paginated_fork.rs b/codex-rs/thread-store/src/local/paginated_fork.rs index b76ec3e282..277ca00edb 100644 --- a/codex-rs/thread-store/src/local/paginated_fork.rs +++ b/codex-rs/thread-store/src/local/paginated_fork.rs @@ -46,17 +46,19 @@ pub(super) async fn prepare( operation: "prepare_fork", }); } - let (history_base, model_context) = lineage + let (history_base, model_context, preview_before_compaction) = lineage .read_with_fallback(store, |lineage| { read_fork_history(store, thread_id, boundary, lineage) }) .await?; - Ok(PreparedFork::with_source_reservation( + let mut fork = PreparedFork::with_source_reservation( thread_id, history_base, Arc::new(model_context), source_reservation, - )) + ); + fork.preview_before_compaction = preview_before_compaction; + Ok(fork) } async fn read_fork_history( @@ -64,7 +66,7 @@ async fn read_fork_history( thread_id: ThreadId, boundary: &ForkBoundary, lineage: RolloutLineage, -) -> ThreadStoreResult<(Option, Vec)> { +) -> ThreadStoreResult<(Option, Vec, Option)> { let source_segment = lineage .segments() .last() @@ -97,8 +99,19 @@ async fn read_fork_history( let history_base = history_base_at_boundary(store, thread_id, boundary, &lineage).await?; drop(source_writer_guard); - let model_context = model_context::load_for_fork(lineage, history_base).await?; - Ok((history_base, model_context)) + let model_context = model_context::load_for_fork(lineage.clone(), history_base).await?; + // A bounded replay starts with session metadata followed by the compaction. Its opening + // user message can only be recovered from the transcript within the same fork boundary. + let preview = if let [RolloutItem::SessionMeta(_), RolloutItem::Compacted(_), ..] = + model_context.as_slice() + && let Some(end) = history_base + { + let prefix = lineage.truncate_at(end).await?; + super::thread_history::first_user_preview(store, &prefix).await? + } else { + None + }; + Ok((history_base, model_context, preview)) } pub(super) async fn history_base_at_boundary( diff --git a/codex-rs/thread-store/src/local/thread_history.rs b/codex-rs/thread-store/src/local/thread_history.rs index 34df8ee4a9..aa8ff7a801 100644 --- a/codex-rs/thread-store/src/local/thread_history.rs +++ b/codex-rs/thread-store/src/local/thread_history.rs @@ -22,6 +22,7 @@ pub(super) use realtime::list_timeline; pub(super) use search::search_thread_occurrences; pub(super) use turn_lookup::find_source_turn; pub(super) use turn_lookup::find_visible_turn; +pub(super) use turn_lookup::first_user_preview; /// A valid complete rollout line with its absolute byte span in durable JSONL. /// diff --git a/codex-rs/thread-store/src/local/thread_history/turn_lookup.rs b/codex-rs/thread-store/src/local/thread_history/turn_lookup.rs index 7fe20d6809..ba37aa0721 100644 --- a/codex-rs/thread-store/src/local/thread_history/turn_lookup.rs +++ b/codex-rs/thread-store/src/local/thread_history/turn_lookup.rs @@ -1,9 +1,16 @@ +use codex_app_server_protocol::ThreadItemPayload; +use codex_app_server_protocol::UserInput; use codex_protocol::ThreadId; +use codex_protocol::items::UserMessageItem; +use codex_protocol::protocol::user_message_preview; +use futures::TryStreamExt; use sqlx::Row; +use super::super::LocalThreadStore; use super::super::rollout_lineage::RolloutLineage; use super::super::rollout_lineage::RolloutLineageSegment; use super::sqlite_integer; +use super::thread_history_error; use crate::ThreadStoreError; use crate::ThreadStoreResult; @@ -18,6 +25,50 @@ pub(in crate::local) struct TurnRow { pub final_agent_item_id: Option, } +pub(in crate::local) async fn first_user_preview( + store: &LocalThreadStore, + prefix: &RolloutLineage, +) -> ThreadStoreResult> { + let pool = store.thread_history_db().await?; + for segment in prefix.for_history_page().await?.segments() { + let mut messages = sqlx::query_scalar::<_, String>( + r#"SELECT item_json FROM thread_items + WHERE thread_id = ?1 AND rollout_ordinal >= ?2 AND rollout_ordinal < ?3 + AND updated_at_ordinal < ?3 + AND (item_type = 'userMessage' + OR (item_type = '' AND json_extract(item_json, '$.type') = 'userMessage')) + ORDER BY rollout_ordinal"#, + ) + .bind(segment.rollout_id().to_string()) + .bind(sqlite_integer(segment.start_ordinal(), "rollout ordinal")?) + .bind( + segment + .end_ordinal() + .map(|end| sqlite_integer(end, "rollout ordinal")) + .transpose()?, + ) + .fetch(pool); + while let Some(json) = messages.try_next().await.map_err(thread_history_error)? { + if let ThreadItemPayload::UserMessage { + id, + client_id, + content, + } = serde_json::from_str(&json).map_err(thread_history_error)? + { + let user = UserMessageItem { + id, + client_id, + content: content.into_iter().map(UserInput::into_core).collect(), + }; + if let Some(preview) = user_message_preview(&user.as_legacy_user_message_event()) { + return Ok(Some(preview)); + } + } + } + } + Ok(None) +} + pub(in crate::local) async fn find_source_turn( pool: &sqlx::SqlitePool, lineage: &RolloutLineage, diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index 31cf522a4e..0c80650162 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -55,7 +55,10 @@ pub(crate) struct PendingThreadMetadataPatch { } impl ThreadMetadataSync { - pub(crate) async fn for_create(params: &CreateThreadParams) -> Self { + pub(crate) async fn for_create( + params: &CreateThreadParams, + inherited_preview: Option<&str>, + ) -> Self { let created_at = Utc::now(); let cwd = params.metadata.cwd.clone().unwrap_or_default(); let git_info = if get_git_repo_root(cwd.as_path()).is_some() { @@ -93,7 +96,8 @@ impl ThreadMetadataSync { Self { thread_id: params.thread_id, cwd_seen: !cwd.as_os_str().is_empty(), - preview_seen: guardian_review, + preview_seen: guardian_review + || inherited_preview.is_some_and(|preview| !preview.is_empty()), first_user_message_seen: guardian_review, title_seen: guardian_review, pending_update: Some(update), @@ -485,7 +489,7 @@ mod tests { #[tokio::test] async fn create_metadata_records_originator_without_project() { let thread_id = ThreadId::new(); - let sync = ThreadMetadataSync::for_create(&CreateThreadParams { + let params = CreateThreadParams { attached_thread: false, creator_user_id: None, creator_account_id: None, @@ -511,8 +515,8 @@ mod tests { model_provider: "test-provider".to_string(), memory_mode: ThreadMemoryMode::Enabled, }, - }) - .await; + }; + let sync = ThreadMetadataSync::for_create(¶ms, /*inherited_preview*/ None).await; let update = sync.take_pending_update().expect("pending metadata update"); assert_eq!( diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 8effae5df6..66c26692df 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -244,6 +244,8 @@ pub struct PreparedFork { pub history_base: Option, /// Bounded model context selected by the requested fork boundary. pub model_context: Arc>, + /// Opening preview from the frozen transcript when compaction omitted it from model context. + pub preview_before_compaction: Option, /// Prevents the source thread from being deleted until the child's history reference is /// finalized. /// @@ -264,6 +266,7 @@ impl PreparedFork { source_thread_id, history_base, model_context, + preview_before_compaction: None, _source_reservation: Some(Box::new(source_reservation)), } } @@ -279,6 +282,7 @@ impl PreparedFork { source_thread_id, history_base, model_context, + preview_before_compaction: None, _source_reservation: None, } }