From a4a0924bb70bd700592a978feb348b7ae690d2cf Mon Sep 17 00:00:00 2001 From: anais Date: Thu, 18 Jun 2026 17:36:36 -0700 Subject: [PATCH] fix codex fork summary feedback --- .../request_processors/thread_processor.rs | 10 +- .../thread_processor_tests.rs | 78 +++++++++ codex-rs/rollout/src/list.rs | 156 ++++++++++++------ codex-rs/rollout/src/tests.rs | 59 +++++++ 4 files changed, 242 insertions(+), 61 deletions(-) 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 e5cb3662dc..dd0975ccb4 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -4494,15 +4494,7 @@ fn rollout_item_opens_implicit_turn(item: &RolloutItem) -> bool { } fn rollout_item_disables_recent_not_loaded_fast_path(item: &RolloutItem) -> bool { - matches!( - item, - RolloutItem::EventMsg( - EventMsg::ThreadRolledBack(_) - | EventMsg::TurnStarted(_) - | EventMsg::TurnComplete(_) - | EventMsg::TurnAborted(_) - ) - ) + matches!(item, RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_))) } fn build_recent_not_loaded_turns_page_response( diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index bf749e1907..6622102715 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -508,6 +508,84 @@ mod thread_processor_behavior_tests { Ok(()) } + #[test] + fn recent_not_loaded_rollout_items_keeps_lifecycle_events() -> Result<()> { + let temp_dir = TempDir::new()?; + let path = temp_dir.path().join("rollout.jsonl"); + let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?; + let contents = compacted_resume_test_rollout_contents(vec![ + compacted_resume_test_session(thread_id), + RolloutItem::EventMsg(EventMsg::TurnStarted( + codex_protocol::protocol::TurnStartedEvent { + turn_id: "turn-a".to_string(), + trace_id: None, + started_at: None, + model_context_window: None, + collaboration_mode_kind: Default::default(), + }, + )), + compacted_resume_test_user_event("first user event"), + compacted_resume_test_agent_event("first agent event"), + RolloutItem::EventMsg(EventMsg::TurnComplete( + codex_protocol::protocol::TurnCompleteEvent { + turn_id: "turn-a".to_string(), + last_agent_message: None, + completed_at: None, + duration_ms: None, + time_to_first_token_ms: None, + }, + )), + RolloutItem::EventMsg(EventMsg::TurnStarted( + codex_protocol::protocol::TurnStartedEvent { + turn_id: "turn-b".to_string(), + trace_id: None, + started_at: None, + model_context_window: None, + collaboration_mode_kind: Default::default(), + }, + )), + compacted_resume_test_user_event("second user event"), + compacted_resume_test_agent_event("second agent event"), + RolloutItem::EventMsg(EventMsg::TurnComplete( + codex_protocol::protocol::TurnCompleteEvent { + turn_id: "turn-b".to_string(), + last_agent_message: None, + completed_at: None, + duration_ms: None, + time_to_first_token_ms: None, + }, + )), + ])?; + std::fs::write(&path, contents)?; + + let recent = read_recent_not_loaded_rollout_items(path.as_path(), /*page_size*/ 1)? + .expect("recent notLoaded suffix"); + assert!( + recent + .items + .iter() + .any(|item| { matches!(item, RolloutItem::EventMsg(EventMsg::TurnComplete(_))) }) + ); + let page = build_recent_not_loaded_turns_page_response( + recent, + codex_app_server_protocol::ThreadStatus::NotLoaded, + Some(1), + ) + .expect("recent notLoaded page should build"); + + assert_eq!(page.data.len(), 1); + assert_eq!(page.data[0].id, "turn-b"); + assert_eq!( + page.data[0].items_view, + codex_app_server_protocol::TurnItemsView::NotLoaded + ); + assert_eq!( + page.data[0].status, + codex_app_server_protocol::TurnStatus::Completed + ); + Ok(()) + } + #[test] fn validate_dynamic_tools_rejects_empty_namespace() { let tools = vec![dynamic_tool( diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index dd31ca2d17..70837c1f54 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -91,6 +91,7 @@ pub type ConversationsPage = ThreadsPage; #[derive(Default)] struct HeadTailSummary { saw_session_meta: bool, + has_fork_parent_rollout_ref: bool, thread_id: Option, first_user_message: Option, preview: Option, @@ -770,9 +771,17 @@ async fn build_thread_item( ) -> Option { // Read head and detect preview-bearing events; goal previews can appear before // the first normal user message. - let summary = read_head_summary(&path, HEAD_RECORD_LIMIT) + let mut summary = read_head_summary(&path, HEAD_RECORD_LIMIT) .await .unwrap_or_default(); + if summary.saw_session_meta && summary.preview.is_none() && summary.has_fork_parent_rollout_ref + { + if let Ok(expanded_summary) = read_expanded_rollout_summary(&path).await { + if expanded_summary.preview.is_some() { + summary = expanded_summary; + } + } + } if !allowed_sources.is_empty() && !summary .source @@ -857,6 +866,22 @@ pub async fn read_thread_item_from_rollout(path: PathBuf) -> Option .await } +async fn read_expanded_rollout_summary(path: &Path) -> io::Result { + let (items, _, _) = super::recorder::RolloutRecorder::load_rollout_items(path).await?; + let mut summary = HeadTailSummary::default(); + let scan_limit = HEAD_RECORD_LIMIT + USER_EVENT_SCAN_LIMIT; + for item in items.iter().take(scan_limit) { + observe_summary_rollout_item(&mut summary, item, /*timestamp*/ None); + if summary.saw_session_meta + && summary.preview.is_some() + && summary.first_user_message.is_some() + { + break; + } + } + Ok(summary) +} + /// Collects immediate subdirectories of `parent`, parses their (string) names with `parse`, /// and returns them sorted descending by the parsed key. async fn collect_dirs_desc(parent: &Path, parse: F) -> io::Result> @@ -1115,61 +1140,18 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result = serde_json::from_str(trimmed); let Ok(rollout_line) = parsed else { continue }; - match rollout_line.item { - RolloutItem::SessionMeta(session_meta_line) => { - if !summary.saw_session_meta { - summary.source = Some(session_meta_line.meta.source.clone()); - summary.parent_thread_id = session_meta_line.meta.parent_thread_id; - summary.agent_nickname = session_meta_line.meta.agent_nickname.clone(); - summary.agent_role = session_meta_line.meta.agent_role.clone(); - summary.model_provider = session_meta_line.meta.model_provider.clone(); - summary.thread_id = Some(session_meta_line.meta.id); - summary.cwd = Some(session_meta_line.meta.cwd.clone()); - summary.git_branch = session_meta_line - .git - .as_ref() - .and_then(|git| git.branch.clone()); - summary.git_sha = session_meta_line - .git - .as_ref() - .and_then(|git| git.commit_hash.as_ref().map(|sha| sha.0.clone())); - summary.git_origin_url = session_meta_line - .git - .as_ref() - .and_then(|git| git.repository_url.clone()); - summary.cli_version = Some(session_meta_line.meta.cli_version); - summary.created_at = Some(session_meta_line.meta.timestamp.clone()); - summary.saw_session_meta = true; - } - } - RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) => { - summary - .created_at - .get_or_insert_with(|| rollout_line.timestamp.clone()); - } - RolloutItem::TurnContext(_) => { - // Not included in `head`; skip. - } - RolloutItem::Compacted(_) => { - // Not included in `head`; skip. - } - RolloutItem::EventMsg(ev) => { - if let Some(preview) = event_msg_preview(&ev) { - if summary.preview.is_none() { - summary.preview = Some(preview.clone()); - } - if let EventMsg::UserMessage(_) = ev - && summary.first_user_message.is_none() - { - summary.first_user_message = Some(preview); - } - } - } - } + observe_summary_rollout_item( + &mut summary, + &rollout_line.item, + Some(&rollout_line.timestamp), + ); if summary.saw_session_meta && summary.preview.is_some() @@ -1182,6 +1164,76 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result, +) { + match item { + RolloutItem::SessionMeta(session_meta_line) => { + if !summary.saw_session_meta { + summary.source = Some(session_meta_line.meta.source.clone()); + summary.parent_thread_id = session_meta_line.meta.parent_thread_id; + summary.agent_nickname = session_meta_line.meta.agent_nickname.clone(); + summary.agent_role = session_meta_line.meta.agent_role.clone(); + summary.model_provider = session_meta_line.meta.model_provider.clone(); + summary.thread_id = Some(session_meta_line.meta.id); + summary.cwd = Some(session_meta_line.meta.cwd.clone()); + summary.git_branch = session_meta_line + .git + .as_ref() + .and_then(|git| git.branch.clone()); + summary.git_sha = session_meta_line + .git + .as_ref() + .and_then(|git| git.commit_hash.as_ref().map(|sha| sha.0.clone())); + summary.git_origin_url = session_meta_line + .git + .as_ref() + .and_then(|git| git.repository_url.clone()); + summary.cli_version = Some(session_meta_line.meta.cli_version.clone()); + summary.created_at = Some(session_meta_line.meta.timestamp.clone()); + summary.saw_session_meta = true; + } + } + RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) => { + if let Some(timestamp) = timestamp { + summary + .created_at + .get_or_insert_with(|| timestamp.to_string()); + } + } + RolloutItem::TurnContext(_) | RolloutItem::Compacted(_) => {} + RolloutItem::EventMsg(event) => { + if let Some(preview) = event_msg_preview(event) { + if summary.preview.is_none() { + summary.preview = Some(preview.clone()); + } + if let EventMsg::UserMessage(_) = event + && summary.first_user_message.is_none() + { + summary.first_user_message = Some(preview); + } + } + } + } +} + +fn rollout_line_has_fork_parent_rollout_ref(line: &str) -> bool { + let Ok(value) = serde_json::from_str::(line) else { + return false; + }; + let payload = value.get("payload").unwrap_or(&value); + payload + .get("fork_parent_rollout_path") + .and_then(serde_json::Value::as_str) + .is_some_and(|path| !path.is_empty()) + && payload + .get("fork_parent_rollout_byte_len") + .and_then(serde_json::Value::as_u64) + .is_some_and(|byte_len| byte_len > 0) +} + /// Read up to `HEAD_RECORD_LIMIT` records from the start of the rollout file at `path`. /// This should be enough to produce a summary including the session meta line. pub async fn read_head_for_summary(path: &Path) -> io::Result> { diff --git a/codex-rs/rollout/src/tests.rs b/codex-rs/rollout/src/tests.rs index 0e3ae912dd..56f84bd392 100644 --- a/codex-rs/rollout/src/tests.rs +++ b/codex-rs/rollout/src/tests.rs @@ -26,6 +26,7 @@ use crate::list::ThreadSortKey; use crate::list::ThreadsPage; use crate::list::get_threads; use crate::list::read_head_for_summary; +use crate::list::read_thread_item_from_rollout; use crate::rollout_date_parts; use anyhow::Result; use codex_protocol::ThreadId; @@ -1348,6 +1349,64 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> { Ok(()) } +#[tokio::test] +async fn read_thread_item_from_cow_child_uses_parent_preview() -> Result<()> { + let temp = TempDir::new().unwrap(); + let home = temp.path(); + + let parent_ts = "2025-06-01T08-00-00"; + let parent_uuid = Uuid::from_u128(42); + write_session_file( + home, + parent_ts, + parent_uuid, + /*num_records*/ 0, + Some(SessionSource::VSCode), + )?; + let parent_path = home + .join("sessions") + .join("2025") + .join("06") + .join("01") + .join(format!("rollout-{parent_ts}-{parent_uuid}.jsonl")); + let parent_byte_len = fs::metadata(&parent_path)?.len(); + + let child_ts = "2025-06-01T09-00-00"; + let child_uuid = Uuid::from_u128(43); + let child_dir = home.join("sessions").join("2025").join("06").join("01"); + fs::create_dir_all(&child_dir)?; + let child_path = child_dir.join(format!("rollout-{child_ts}-{child_uuid}.jsonl")); + let child_meta = serde_json::json!({ + "timestamp": child_ts, + "type": "session_meta", + "payload": { + "id": child_uuid, + "forked_from_id": parent_uuid, + "timestamp": child_ts, + "cwd": ".", + "originator": "test_originator", + "cli_version": "test_version", + "base_instructions": null, + "source": "vscode", + "model_provider": TEST_PROVIDER, + "fork_parent_rollout_path": parent_path.to_string_lossy(), + "fork_parent_rollout_byte_len": parent_byte_len, + } + }); + fs::write(&child_path, format!("{child_meta}\n"))?; + + let item = read_thread_item_from_rollout(child_path.clone()) + .await + .expect("child summary should expand parent preview"); + + assert_eq!(item.path, child_path); + assert_eq!(item.thread_id, Some(thread_id_from_uuid(child_uuid))); + assert_eq!(item.preview.as_deref(), Some("Hello from user")); + assert_eq!(item.first_user_message.as_deref(), Some("Hello from user")); + assert_eq!(item.model_provider.as_deref(), Some(TEST_PROVIDER)); + Ok(()) +} + #[tokio::test] async fn test_timestamp_only_cursor_skips_same_second_filesystem_ties() { let temp = TempDir::new().unwrap();