From 9f0273b74b7ca3f2e70daafde5fe9573c2d2dc56 Mon Sep 17 00:00:00 2001 From: Adam Van Ymeren Date: Mon, 15 Jun 2026 13:34:44 -0700 Subject: [PATCH 1/2] [app-server] expose stored thread views --- codex-rs/app-server/src/filters.rs | 158 ------- codex-rs/app-server/src/lib.rs | 2 +- codex-rs/app-server/src/request_processors.rs | 20 +- .../request_processors/thread_processor.rs | 285 +----------- .../src/request_processors/thread_summary.rs | 31 -- codex-rs/app-server/src/thread_views.rs | 434 ++++++++++++++++++ codex-rs/app-server/src/thread_views_tests.rs | 200 ++++++++ 7 files changed, 664 insertions(+), 466 deletions(-) delete mode 100644 codex-rs/app-server/src/filters.rs create mode 100644 codex-rs/app-server/src/thread_views.rs create mode 100644 codex-rs/app-server/src/thread_views_tests.rs diff --git a/codex-rs/app-server/src/filters.rs b/codex-rs/app-server/src/filters.rs deleted file mode 100644 index 20608d93cd25..000000000000 --- a/codex-rs/app-server/src/filters.rs +++ /dev/null @@ -1,158 +0,0 @@ -use codex_app_server_protocol::ThreadSourceKind; -use codex_core::INTERACTIVE_SESSION_SOURCES; -use codex_protocol::protocol::SessionSource as CoreSessionSource; -use codex_protocol::protocol::SubAgentSource as CoreSubAgentSource; - -pub(crate) fn compute_source_filters( - source_kinds: Option>, -) -> (Vec, Option>) { - let Some(source_kinds) = source_kinds else { - return (INTERACTIVE_SESSION_SOURCES.to_vec(), None); - }; - - if source_kinds.is_empty() { - return (INTERACTIVE_SESSION_SOURCES.to_vec(), None); - } - - let requires_post_filter = source_kinds.iter().any(|kind| { - matches!( - kind, - ThreadSourceKind::Exec - | ThreadSourceKind::AppServer - | ThreadSourceKind::SubAgent - | ThreadSourceKind::SubAgentReview - | ThreadSourceKind::SubAgentCompact - | ThreadSourceKind::SubAgentThreadSpawn - | ThreadSourceKind::SubAgentOther - | ThreadSourceKind::Unknown - ) - }); - - if requires_post_filter { - (Vec::new(), Some(source_kinds)) - } else { - let interactive_sources = source_kinds - .iter() - .filter_map(|kind| match kind { - ThreadSourceKind::Cli => Some(CoreSessionSource::Cli), - ThreadSourceKind::VsCode => Some(CoreSessionSource::VSCode), - ThreadSourceKind::Exec - | ThreadSourceKind::AppServer - | ThreadSourceKind::SubAgent - | ThreadSourceKind::SubAgentReview - | ThreadSourceKind::SubAgentCompact - | ThreadSourceKind::SubAgentThreadSpawn - | ThreadSourceKind::SubAgentOther - | ThreadSourceKind::Unknown => None, - }) - .collect::>(); - (interactive_sources, Some(source_kinds)) - } -} - -pub(crate) fn source_kind_matches(source: &CoreSessionSource, filter: &[ThreadSourceKind]) -> bool { - filter.iter().any(|kind| match kind { - ThreadSourceKind::Cli => matches!(source, CoreSessionSource::Cli), - ThreadSourceKind::VsCode => matches!(source, CoreSessionSource::VSCode), - ThreadSourceKind::Exec => matches!(source, CoreSessionSource::Exec), - ThreadSourceKind::AppServer => matches!(source, CoreSessionSource::Mcp), - ThreadSourceKind::SubAgent => matches!(source, CoreSessionSource::SubAgent(_)), - ThreadSourceKind::SubAgentReview => { - matches!( - source, - CoreSessionSource::SubAgent(CoreSubAgentSource::Review) - ) - } - ThreadSourceKind::SubAgentCompact => { - matches!( - source, - CoreSessionSource::SubAgent(CoreSubAgentSource::Compact) - ) - } - ThreadSourceKind::SubAgentThreadSpawn => matches!( - source, - CoreSessionSource::SubAgent(CoreSubAgentSource::ThreadSpawn { .. }) - ), - ThreadSourceKind::SubAgentOther => matches!( - source, - CoreSessionSource::SubAgent(CoreSubAgentSource::Other(_)) - ), - ThreadSourceKind::Unknown => matches!(source, CoreSessionSource::Unknown), - }) -} - -#[cfg(test)] -mod tests { - use super::*; - use codex_protocol::ThreadId; - use pretty_assertions::assert_eq; - use uuid::Uuid; - - #[test] - fn compute_source_filters_defaults_to_interactive_sources() { - let (allowed_sources, filter) = compute_source_filters(/*source_kinds*/ None); - - assert_eq!(allowed_sources, INTERACTIVE_SESSION_SOURCES.to_vec()); - assert_eq!(filter, None); - } - - #[test] - fn compute_source_filters_empty_means_interactive_sources() { - let (allowed_sources, filter) = compute_source_filters(Some(Vec::new())); - - assert_eq!(allowed_sources, INTERACTIVE_SESSION_SOURCES.to_vec()); - assert_eq!(filter, None); - } - - #[test] - fn compute_source_filters_interactive_only_skips_post_filtering() { - let source_kinds = vec![ThreadSourceKind::Cli, ThreadSourceKind::VsCode]; - let (allowed_sources, filter) = compute_source_filters(Some(source_kinds.clone())); - - assert_eq!( - allowed_sources, - vec![CoreSessionSource::Cli, CoreSessionSource::VSCode] - ); - assert_eq!(filter, Some(source_kinds)); - } - - #[test] - fn compute_source_filters_subagent_variant_requires_post_filtering() { - let source_kinds = vec![ThreadSourceKind::SubAgentReview]; - let (allowed_sources, filter) = compute_source_filters(Some(source_kinds.clone())); - - assert_eq!(allowed_sources, Vec::new()); - assert_eq!(filter, Some(source_kinds)); - } - - #[test] - fn source_kind_matches_distinguishes_subagent_variants() { - let parent_thread_id = - ThreadId::from_string(&Uuid::new_v4().to_string()).expect("valid thread id"); - let review = CoreSessionSource::SubAgent(CoreSubAgentSource::Review); - let spawn = CoreSessionSource::SubAgent(CoreSubAgentSource::ThreadSpawn { - parent_thread_id, - depth: 1, - agent_path: None, - agent_nickname: None, - agent_role: None, - }); - - assert!(source_kind_matches( - &review, - &[ThreadSourceKind::SubAgentReview] - )); - assert!(!source_kind_matches( - &review, - &[ThreadSourceKind::SubAgentThreadSpawn] - )); - assert!(source_kind_matches( - &spawn, - &[ThreadSourceKind::SubAgentThreadSpawn] - )); - assert!(!source_kind_matches( - &spawn, - &[ThreadSourceKind::SubAgentReview] - )); - } -} diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index db73217a58ce..395da8390823 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -96,7 +96,6 @@ mod current_time; mod dynamic_tools; mod error_code; mod extensions; -mod filters; mod fs_watch; mod fuzzy_file_search; mod image_url; @@ -113,6 +112,7 @@ mod server_request_error; mod skills_watcher; mod thread_state; mod thread_status; +pub mod thread_views; mod transport; pub use crate::error_code::INPUT_TOO_LARGE_ERROR_CODE; diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index d73a41706caf..0fe8efec4e7c 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -78,6 +78,7 @@ use codex_app_server_protocol::GetConversationSummaryResponse; use codex_app_server_protocol::GetWorkspaceMessagesResponse; use codex_app_server_protocol::GitDiffToRemoteParams; use codex_app_server_protocol::GitDiffToRemoteResponse; +#[cfg(test)] use codex_app_server_protocol::GitInfo as ApiGitInfo; use codex_app_server_protocol::HookMetadata; use codex_app_server_protocol::HooksListParams; @@ -204,7 +205,6 @@ use codex_app_server_protocol::ThreadGoalSetParams; use codex_app_server_protocol::ThreadGoalSetResponse; use codex_app_server_protocol::ThreadGoalStatus; use codex_app_server_protocol::ThreadGoalUpdatedNotification; -use codex_app_server_protocol::ThreadHistoryBuilder; use codex_app_server_protocol::ThreadIncrementElicitationParams; use codex_app_server_protocol::ThreadIncrementElicitationResponse; use codex_app_server_protocol::ThreadInjectItemsParams; @@ -429,7 +429,6 @@ use codex_protocol::protocol::W3cTraceContext; 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; -use codex_rollout::is_persisted_rollout_item; use codex_rollout::state_db::StateDbHandle; use codex_rollout::state_db::reconcile_rollout; use codex_state::ThreadMetadata; @@ -530,12 +529,12 @@ pub(crate) use windows_sandbox_processor::WindowsSandboxRequestProcessor; use crate::error_code::internal_error; use crate::error_code::invalid_request; -use crate::filters::compute_source_filters; -use crate::filters::source_kind_matches; use crate::thread_state::ConnectionCapabilities; use crate::thread_state::ThreadListenerCommand; use crate::thread_state::ThreadState; use crate::thread_state::ThreadStateManager; +use crate::thread_views::ThreadSourceFilter; +use crate::thread_views::with_thread_spawn_agent_metadata; use token_usage_replay::latest_token_usage_turn_id_from_rollout_items; use token_usage_replay::send_thread_token_usage_update_to_connection; @@ -603,20 +602,11 @@ use self::thread_resume_redaction::*; use self::thread_summary::*; pub(crate) use self::thread_lifecycle::populate_thread_turns_from_history; -pub(crate) use self::thread_processor::thread_from_stored_thread; #[cfg(test)] pub(crate) use self::thread_summary::read_summary_from_rollout; #[cfg(test)] 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 { - let mut builder = ThreadHistoryBuilder::new(); - for item in items { - if is_persisted_rollout_item(item) { - builder.handle_rollout_item(item); - } - } - builder.finish() -} +pub(crate) use crate::thread_views::build_api_turns_from_rollout_items; +pub(crate) use crate::thread_views::from_stored_thread_with_history as thread_from_stored_thread; 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 7242e816beda..cd6e75f5b7a3 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2002,7 +2002,7 @@ impl ThreadRequestProcessor { ThreadSortKey::RecencyAt => StoreThreadSortKey::RecencyAt, }; let store_sort_direction = sort_direction.unwrap_or(SortDirection::Desc); - let (allowed_sources, source_kind_filter) = compute_source_filters(source_kinds); + let source_filter = ThreadSourceFilter::new(source_kinds); let mut cursor_obj = cursor; let mut last_cursor = cursor_obj.clone(); let mut remaining = requested_page_size; @@ -2020,7 +2020,7 @@ impl ThreadRequestProcessor { SortDirection::Asc => StoreSortDirection::Asc, SortDirection::Desc => StoreSortDirection::Desc, }, - allowed_sources: allowed_sources.clone(), + allowed_sources: source_filter.store_sources().to_vec(), archived: archived.unwrap_or(false), search_term: search_term.clone(), }) @@ -2033,10 +2033,7 @@ impl ThreadRequestProcessor { result.thread.agent_nickname.clone(), result.thread.agent_role.clone(), ); - if source_kind_filter - .as_ref() - .is_none_or(|filter| source_kind_matches(&source, filter)) - { + if source_filter.matches(&source) { search_results.push(result); if search_results.len() >= requested_page_size { break; @@ -2106,46 +2103,14 @@ impl ThreadRequestProcessor { params: ThreadLoadedListParams, ) -> Result { let ThreadLoadedListParams { cursor, limit } = params; - let mut data: Vec = self + let data: Vec = self .thread_manager .list_thread_ids() .await .into_iter() .map(|thread_id| thread_id.to_string()) .collect(); - - if data.is_empty() { - return Ok(ThreadLoadedListResponse { - data, - next_cursor: None, - }); - } - - data.sort(); - let total = data.len(); - let start = match cursor { - Some(cursor) => { - let cursor = match ThreadId::from_string(&cursor) { - Ok(id) => id.to_string(), - Err(_) => return Err(invalid_request(format!("invalid cursor: {cursor}"))), - }; - match data.binary_search(&cursor) { - Ok(idx) => idx + 1, - Err(idx) => idx, - } - } - None => 0, - }; - - let effective_limit = limit.unwrap_or(total as u32).max(1) as usize; - let end = start.saturating_add(effective_limit).min(total); - let page = data[start..end].to_vec(); - let next_cursor = page.last().filter(|_| end < total).cloned(); - - Ok(ThreadLoadedListResponse { - data: page, - next_cursor, - }) + crate::thread_views::paginate_loaded_thread_ids(data, cursor.as_deref(), limit) } async fn thread_read_response_inner( @@ -3682,13 +3647,15 @@ impl ThreadRequestProcessor { None if relation_filter.is_some() => None, None => Some(vec![self.config.model_provider_id.clone()]), }; - let (allowed_sources_vec, source_kind_filter) = - if relation_filter.is_some() && source_kinds.is_none() { - (Vec::new(), None) - } else { - compute_source_filters(source_kinds) - }; - let allowed_sources = allowed_sources_vec.as_slice(); + let source_filter = if relation_filter.is_some() && source_kinds.is_none() { + None + } else { + Some(ThreadSourceFilter::new(source_kinds)) + }; + let allowed_sources = source_filter + .as_ref() + .map(ThreadSourceFilter::store_sources) + .unwrap_or_default(); let store_sort_direction = match sort_direction { SortDirection::Asc => StoreSortDirection::Asc, SortDirection::Desc => StoreSortDirection::Desc, @@ -3721,9 +3688,9 @@ impl ThreadRequestProcessor { it.agent_nickname.clone(), it.agent_role.clone(), ); - if source_kind_filter + if source_filter .as_ref() - .is_none_or(|filter| source_kind_matches(&source, filter)) + .is_none_or(|filter| filter.matches(&source)) && cwd_filters.as_ref().is_none_or(|expected_cwds| { expected_cwds.iter().any(|expected_cwd| { path_utils::paths_match_after_normalization(&it.cwd, expected_cwd) @@ -3772,11 +3739,8 @@ fn xcode_26_4_mcp_elicitations_auto_deny( && client_version.is_some_and(|version| version.starts_with("26.4")) } -const THREAD_TURNS_DEFAULT_LIMIT: usize = 25; -const THREAD_TURNS_MAX_LIMIT: usize = 100; const THREAD_ITEMS_DEFAULT_LIMIT: usize = 25; const THREAD_ITEMS_MAX_LIMIT: usize = 100; - fn thread_backwards_cursor_for_sort_key( thread: &StoredThread, sort_key: StoreThreadSortKey, @@ -3796,113 +3760,6 @@ fn thread_backwards_cursor_for_sort_key( Some(timestamp.to_rfc3339_opts(SecondsFormat::Millis, true)) } -struct ThreadTurnsPage { - pub(super) turns: Vec, - pub(super) next_cursor: Option, - pub(super) backwards_cursor: Option, -} - -#[derive(serde::Serialize, serde::Deserialize)] -#[serde(rename_all = "camelCase")] -struct ThreadTurnsCursor { - turn_id: String, - include_anchor: bool, -} - -fn paginate_thread_turns( - turns: Vec, - cursor: Option<&str>, - limit: Option, - sort_direction: SortDirection, -) -> Result { - if turns.is_empty() { - return Ok(ThreadTurnsPage { - turns: Vec::new(), - next_cursor: None, - backwards_cursor: None, - }); - } - - let anchor = cursor.map(parse_thread_turns_cursor).transpose()?; - let page_size = limit - .map(|value| value as usize) - .unwrap_or(THREAD_TURNS_DEFAULT_LIMIT) - .clamp(1, THREAD_TURNS_MAX_LIMIT); - - let anchor_index = anchor - .as_ref() - .and_then(|anchor| turns.iter().position(|turn| turn.id == anchor.turn_id)); - if anchor.is_some() && anchor_index.is_none() { - return Err(invalid_request( - "invalid cursor: anchor turn is no longer present", - )); - } - - let mut keyed_turns: Vec<_> = turns.into_iter().enumerate().collect(); - match sort_direction { - SortDirection::Asc => { - if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) { - keyed_turns.retain(|(index, _)| { - if anchor.include_anchor { - *index >= anchor_index - } else { - *index > anchor_index - } - }); - } - } - SortDirection::Desc => { - keyed_turns.reverse(); - if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) { - keyed_turns.retain(|(index, _)| { - if anchor.include_anchor { - *index <= anchor_index - } else { - *index < anchor_index - } - }); - } - } - } - - let more_turns_available = keyed_turns.len() > page_size; - keyed_turns.truncate(page_size); - let backwards_cursor = keyed_turns - .first() - .map(|(_, turn)| serialize_thread_turns_cursor(&turn.id, /*include_anchor*/ true)) - .transpose()?; - let next_cursor = if more_turns_available { - keyed_turns - .last() - .map(|(_, turn)| serialize_thread_turns_cursor(&turn.id, /*include_anchor*/ false)) - .transpose()? - } else { - None - }; - let turns = keyed_turns.into_iter().map(|(_, turn)| turn).collect(); - - Ok(ThreadTurnsPage { - turns, - next_cursor, - backwards_cursor, - }) -} - -fn serialize_thread_turns_cursor( - turn_id: &str, - include_anchor: bool, -) -> Result { - serde_json::to_string(&ThreadTurnsCursor { - turn_id: turn_id.to_string(), - include_anchor, - }) - .map_err(|err| internal_error(format!("failed to serialize cursor: {err}"))) -} - -fn parse_thread_turns_cursor(cursor: &str) -> Result { - serde_json::from_str(cursor).map_err(|_| invalid_request(format!("invalid cursor: {cursor}"))) -} - struct ThreadTurnsPageOptions<'a> { cursor: Option<&'a str>, limit: Option, @@ -3917,19 +3774,19 @@ fn build_thread_turns_page_response( active_turn: Option, options: ThreadTurnsPageOptions<'_>, ) -> Result { - let mut turns = reconstruct_thread_turns_for_turns_list( + let turns = reconstruct_thread_turns_for_turns_list( items, loaded_status, has_live_running_thread, active_turn, ); - apply_thread_turns_items_view(&mut turns, options.items_view); - let page = paginate_thread_turns(turns, options.cursor, options.limit, options.sort_direction)?; - Ok(ThreadTurnsListResponse { - data: page.turns, - next_cursor: page.next_cursor, - backwards_cursor: page.backwards_cursor, - }) + crate::thread_views::paginate_turns( + turns, + options.cursor, + options.limit, + options.sort_direction, + options.items_view, + ) } pub(super) fn build_thread_resume_initial_turns_page( @@ -3954,44 +3811,6 @@ pub(super) fn build_thread_resume_initial_turns_page( .map(Into::into) } -fn apply_thread_turns_items_view(turns: &mut [Turn], items_view: TurnItemsView) { - for turn in turns { - match items_view { - TurnItemsView::NotLoaded => { - turn.items.clear(); - turn.items_view = TurnItemsView::NotLoaded; - } - TurnItemsView::Summary => { - let first_user_message = turn - .items - .iter() - .find(|item| matches!(item, ThreadItem::UserMessage { .. })) - .cloned(); - let final_agent_message = turn - .items - .iter() - .rev() - .find(|item| matches!(item, ThreadItem::AgentMessage { .. })) - .cloned(); - turn.items = match (first_user_message, final_agent_message) { - (Some(user_message), Some(agent_message)) - if user_message.id() != agent_message.id() => - { - vec![user_message, agent_message] - } - (Some(user_message), _) => vec![user_message], - (None, Some(agent_message)) => vec![agent_message], - (None, None) => Vec::new(), - }; - turn.items_view = TurnItemsView::Summary; - } - TurnItemsView::Full => { - turn.items_view = TurnItemsView::Full; - } - } - } -} - fn reconstruct_thread_turns_for_turns_list( items: &[RolloutItem], loaded_status: ThreadStatus, @@ -4191,62 +4010,6 @@ fn set_thread_name_from_title(thread: &mut Thread, title: String) { thread.name = Some(title); } -pub(crate) fn thread_from_stored_thread( - thread: StoredThread, - fallback_provider: &str, - fallback_cwd: &AbsolutePathBuf, -) -> (Thread, Option) { - let path = thread.rollout_path; - let git_info = thread.git_info.map(|info| ApiGitInfo { - sha: info.commit_hash.map(|sha| sha.0), - branch: info.branch, - origin_url: info.repository_url, - }); - let cwd = AbsolutePathBuf::relative_to_current_dir(path_utils::normalize_for_native_workdir( - thread.cwd, - )) - .unwrap_or_else(|err| { - warn!("failed to normalize thread cwd while reading stored thread: {err}"); - fallback_cwd.clone() - }); - let source = with_thread_spawn_agent_metadata( - thread.source, - thread.agent_nickname.clone(), - thread.agent_role.clone(), - ); - let history = thread.history; - let thread_id = thread.thread_id.to_string(); - let thread = Thread { - id: thread_id.clone(), - extra: None, - session_id: thread_id, - forked_from_id: thread.forked_from_id.map(|id| id.to_string()), - parent_thread_id: thread.parent_thread_id.map(|id| id.to_string()), - preview: thread.preview, - ephemeral: false, - model_provider: if thread.model_provider.is_empty() { - fallback_provider.to_string() - } else { - thread.model_provider - }, - created_at: thread.created_at.timestamp(), - updated_at: thread.updated_at.timestamp(), - recency_at: Some(thread.recency_at.timestamp()), - status: ThreadStatus::NotLoaded, - path, - cwd, - cli_version: thread.cli_version, - agent_nickname: source.get_nickname(), - agent_role: source.get_agent_role(), - source: source.into(), - thread_source: thread.thread_source.map(Into::into), - git_info, - name: thread.name, - turns: Vec::new(), - }; - (thread, history) -} - fn summary_from_stored_thread( thread: StoredThread, fallback_provider: &str, 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 2bacaf644680..26eb1da6521c 100644 --- a/codex-rs/app-server/src/request_processors/thread_summary.rs +++ b/codex-rs/app-server/src/request_processors/thread_summary.rs @@ -138,37 +138,6 @@ fn map_git_info(git_info: &CoreGitInfo) -> ConversationGitInfo { } } -pub(super) fn with_thread_spawn_agent_metadata( - source: codex_protocol::protocol::SessionSource, - agent_nickname: Option, - agent_role: Option, -) -> codex_protocol::protocol::SessionSource { - if agent_nickname.is_none() && agent_role.is_none() { - return source; - } - - match source { - codex_protocol::protocol::SessionSource::SubAgent( - codex_protocol::protocol::SubAgentSource::ThreadSpawn { - parent_thread_id, - depth, - agent_path, - agent_nickname: existing_agent_nickname, - agent_role: existing_agent_role, - }, - ) => codex_protocol::protocol::SessionSource::SubAgent( - codex_protocol::protocol::SubAgentSource::ThreadSpawn { - parent_thread_id, - depth, - agent_path, - agent_nickname: agent_nickname.or(existing_agent_nickname), - agent_role: agent_role.or(existing_agent_role), - }, - ), - _ => source, - } -} - pub(crate) fn thread_response_active_permission_profile( active_permission_profile: Option, ) -> Option { diff --git a/codex-rs/app-server/src/thread_views.rs b/codex-rs/app-server/src/thread_views.rs new file mode 100644 index 000000000000..6d5fcd7620ce --- /dev/null +++ b/codex-rs/app-server/src/thread_views.rs @@ -0,0 +1,434 @@ +//! Projection, filtering, and pagination helpers for stored thread data. + +use codex_app_server_protocol::GitInfo; +use codex_app_server_protocol::JSONRPCErrorError; +use codex_app_server_protocol::SortDirection; +use codex_app_server_protocol::Thread; +use codex_app_server_protocol::ThreadHistoryBuilder; +use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadLoadedListResponse; +use codex_app_server_protocol::ThreadSourceKind; +use codex_app_server_protocol::ThreadStatus; +use codex_app_server_protocol::ThreadTurnsListResponse; +use codex_app_server_protocol::Turn; +use codex_app_server_protocol::TurnItemsView; +use codex_core::INTERACTIVE_SESSION_SOURCES; +use codex_core::path_utils; +use codex_protocol::ThreadId; +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; +use codex_rollout::is_persisted_rollout_item; +use codex_thread_store::StoredThread; +use codex_thread_store::StoredThreadHistory; +use codex_utils_absolute_path::AbsolutePathBuf; +use tracing::warn; + +use crate::error_code::internal_error; +use crate::error_code::invalid_request; + +const THREAD_TURNS_DEFAULT_LIMIT: usize = 25; +const THREAD_TURNS_MAX_LIMIT: usize = 100; + +/// Projects a stored thread into the app-server representation. +/// +/// When stored history is present, the returned thread contains reconstructed +/// turns. Runtime-only status and active-turn state are intentionally omitted. +pub fn from_stored_thread( + stored_thread: StoredThread, + fallback_provider: &str, + fallback_cwd: &AbsolutePathBuf, +) -> Thread { + let (mut thread, history) = + from_stored_thread_with_history(stored_thread, fallback_provider, fallback_cwd); + if let Some(history) = history { + thread.turns = build_api_turns_from_rollout_items(&history.items); + } + thread +} + +pub(crate) fn from_stored_thread_with_history( + stored_thread: StoredThread, + fallback_provider: &str, + fallback_cwd: &AbsolutePathBuf, +) -> (Thread, Option) { + let path = stored_thread.rollout_path; + let git_info = stored_thread.git_info.map(|info| GitInfo { + sha: info.commit_hash.map(|sha| sha.0), + branch: info.branch, + origin_url: info.repository_url, + }); + let cwd = AbsolutePathBuf::relative_to_current_dir(path_utils::normalize_for_native_workdir( + stored_thread.cwd, + )) + .unwrap_or_else(|err| { + warn!("failed to normalize thread cwd while reading stored thread: {err}"); + fallback_cwd.clone() + }); + let source = with_thread_spawn_agent_metadata( + stored_thread.source, + stored_thread.agent_nickname.clone(), + stored_thread.agent_role.clone(), + ); + let history = stored_thread.history; + let thread_id = stored_thread.thread_id.to_string(); + let thread = Thread { + id: thread_id.clone(), + extra: None, + session_id: thread_id, + forked_from_id: stored_thread.forked_from_id.map(|id| id.to_string()), + parent_thread_id: stored_thread.parent_thread_id.map(|id| id.to_string()), + preview: stored_thread.preview, + ephemeral: false, + model_provider: if stored_thread.model_provider.is_empty() { + fallback_provider.to_string() + } else { + stored_thread.model_provider + }, + created_at: stored_thread.created_at.timestamp(), + updated_at: stored_thread.updated_at.timestamp(), + recency_at: Some(stored_thread.recency_at.timestamp()), + status: ThreadStatus::NotLoaded, + path, + cwd, + cli_version: stored_thread.cli_version, + agent_nickname: source.get_nickname(), + agent_role: source.get_agent_role(), + source: source.into(), + thread_source: stored_thread.thread_source.map(Into::into), + git_info, + name: stored_thread.name, + turns: Vec::new(), + }; + (thread, history) +} + +pub(crate) fn build_api_turns_from_rollout_items(items: &[RolloutItem]) -> Vec { + let mut builder = ThreadHistoryBuilder::new(); + for item in items { + if is_persisted_rollout_item(item) { + builder.handle_rollout_item(item); + } + } + builder.finish() +} + +/// Source filtering policy shared by thread list and search projections. +#[derive(Clone, Debug)] +pub struct ThreadSourceFilter { + store_sources: Vec, + source_kinds: Option>, +} + +impl ThreadSourceFilter { + /// Creates a filter from requested app-server source kinds. + /// + /// An absent or empty request selects interactive sources. An empty value + /// from [`store_sources`](Self::store_sources) means the backing store must + /// query all sources and let [`matches`](Self::matches) apply the precise + /// classification afterwards. + pub fn new(source_kinds: Option>) -> Self { + let Some(source_kinds) = source_kinds.filter(|source_kinds| !source_kinds.is_empty()) + else { + return Self { + store_sources: INTERACTIVE_SESSION_SOURCES.to_vec(), + source_kinds: None, + }; + }; + + let requires_post_filter = source_kinds.iter().any(|kind| { + matches!( + kind, + ThreadSourceKind::Exec + | ThreadSourceKind::AppServer + | ThreadSourceKind::SubAgent + | ThreadSourceKind::SubAgentReview + | ThreadSourceKind::SubAgentCompact + | ThreadSourceKind::SubAgentThreadSpawn + | ThreadSourceKind::SubAgentOther + | ThreadSourceKind::Unknown + ) + }); + let store_sources = if requires_post_filter { + Vec::new() + } else { + source_kinds + .iter() + .filter_map(|kind| match kind { + ThreadSourceKind::Cli => Some(SessionSource::Cli), + ThreadSourceKind::VsCode => Some(SessionSource::VSCode), + ThreadSourceKind::Exec + | ThreadSourceKind::AppServer + | ThreadSourceKind::SubAgent + | ThreadSourceKind::SubAgentReview + | ThreadSourceKind::SubAgentCompact + | ThreadSourceKind::SubAgentThreadSpawn + | ThreadSourceKind::SubAgentOther + | ThreadSourceKind::Unknown => None, + }) + .collect() + }; + Self { + store_sources, + source_kinds: Some(source_kinds), + } + } + + /// Returns the coarse source set suitable for a backing-store query. + pub fn store_sources(&self) -> &[SessionSource] { + &self.store_sources + } + + /// Returns whether a projected source satisfies the precise filter. + pub fn matches(&self, source: &SessionSource) -> bool { + self.source_kinds.as_ref().is_none_or(|source_kinds| { + source_kinds.iter().any(|kind| match kind { + ThreadSourceKind::Cli => matches!(source, SessionSource::Cli), + ThreadSourceKind::VsCode => matches!(source, SessionSource::VSCode), + ThreadSourceKind::Exec => matches!(source, SessionSource::Exec), + ThreadSourceKind::AppServer => matches!(source, SessionSource::Mcp), + ThreadSourceKind::SubAgent => matches!(source, SessionSource::SubAgent(_)), + ThreadSourceKind::SubAgentReview => { + matches!(source, SessionSource::SubAgent(SubAgentSource::Review)) + } + ThreadSourceKind::SubAgentCompact => { + matches!(source, SessionSource::SubAgent(SubAgentSource::Compact)) + } + ThreadSourceKind::SubAgentThreadSpawn => matches!( + source, + SessionSource::SubAgent(SubAgentSource::ThreadSpawn { .. }) + ), + ThreadSourceKind::SubAgentOther => { + matches!(source, SessionSource::SubAgent(SubAgentSource::Other(_))) + } + ThreadSourceKind::Unknown => matches!(source, SessionSource::Unknown), + }) + }) + } +} + +/// Sorts and paginates a snapshot of loaded thread IDs. +pub fn paginate_loaded_thread_ids( + mut data: Vec, + cursor: Option<&str>, + limit: Option, +) -> Result { + if data.is_empty() { + return Ok(ThreadLoadedListResponse { + data, + next_cursor: None, + }); + } + + data.sort(); + let total = data.len(); + let start = match cursor { + Some(cursor) => { + let cursor = ThreadId::from_string(cursor) + .map_err(|_| invalid_request(format!("invalid cursor: {cursor}")))? + .to_string(); + match data.binary_search(&cursor) { + Ok(index) => index + 1, + Err(index) => index, + } + } + None => 0, + }; + let limit = limit.unwrap_or(total as u32).max(1) as usize; + let end = start.saturating_add(limit).min(total); + let data = data.into_iter().skip(start).take(limit).collect::>(); + let next_cursor = data.last().filter(|_| end < total).cloned(); + Ok(ThreadLoadedListResponse { data, next_cursor }) +} + +/// Applies an item view and paginates reconstructed turns. +pub fn paginate_turns( + mut turns: Vec, + cursor: Option<&str>, + limit: Option, + sort_direction: SortDirection, + items_view: TurnItemsView, +) -> Result { + apply_turn_items_view(&mut turns, items_view); + let page = paginate_turns_inner(turns, cursor, limit, sort_direction)?; + Ok(ThreadTurnsListResponse { + data: page.turns, + next_cursor: page.next_cursor, + backwards_cursor: page.backwards_cursor, + }) +} + +struct TurnsPage { + turns: Vec, + next_cursor: Option, + backwards_cursor: Option, +} + +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct TurnsCursor { + turn_id: String, + include_anchor: bool, +} + +fn paginate_turns_inner( + turns: Vec, + cursor: Option<&str>, + limit: Option, + sort_direction: SortDirection, +) -> Result { + if turns.is_empty() { + return Ok(TurnsPage { + turns: Vec::new(), + next_cursor: None, + backwards_cursor: None, + }); + } + + let anchor = cursor.map(parse_turns_cursor).transpose()?; + let page_size = limit + .map(|value| value as usize) + .unwrap_or(THREAD_TURNS_DEFAULT_LIMIT) + .clamp(1, THREAD_TURNS_MAX_LIMIT); + let anchor_index = anchor + .as_ref() + .and_then(|anchor| turns.iter().position(|turn| turn.id == anchor.turn_id)); + if anchor.is_some() && anchor_index.is_none() { + return Err(invalid_request( + "invalid cursor: anchor turn is no longer present", + )); + } + + let mut keyed_turns: Vec<_> = turns.into_iter().enumerate().collect(); + match sort_direction { + SortDirection::Asc => { + if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) { + keyed_turns.retain(|(index, _)| { + if anchor.include_anchor { + *index >= anchor_index + } else { + *index > anchor_index + } + }); + } + } + SortDirection::Desc => { + keyed_turns.reverse(); + if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) { + keyed_turns.retain(|(index, _)| { + if anchor.include_anchor { + *index <= anchor_index + } else { + *index < anchor_index + } + }); + } + } + } + + let more_turns_available = keyed_turns.len() > page_size; + keyed_turns.truncate(page_size); + let backwards_cursor = keyed_turns + .first() + .map(|(_, turn)| serialize_turns_cursor(&turn.id, /*include_anchor*/ true)) + .transpose()?; + let next_cursor = if more_turns_available { + keyed_turns + .last() + .map(|(_, turn)| serialize_turns_cursor(&turn.id, /*include_anchor*/ false)) + .transpose()? + } else { + None + }; + let turns = keyed_turns.into_iter().map(|(_, turn)| turn).collect(); + + Ok(TurnsPage { + turns, + next_cursor, + backwards_cursor, + }) +} + +fn serialize_turns_cursor( + turn_id: &str, + include_anchor: bool, +) -> Result { + serde_json::to_string(&TurnsCursor { + turn_id: turn_id.to_string(), + include_anchor, + }) + .map_err(|err| internal_error(format!("failed to serialize cursor: {err}"))) +} + +fn parse_turns_cursor(cursor: &str) -> Result { + serde_json::from_str(cursor).map_err(|_| invalid_request(format!("invalid cursor: {cursor}"))) +} + +fn apply_turn_items_view(turns: &mut [Turn], items_view: TurnItemsView) { + for turn in turns { + match items_view { + TurnItemsView::NotLoaded => { + turn.items.clear(); + turn.items_view = TurnItemsView::NotLoaded; + } + TurnItemsView::Summary => { + let first_user_message = turn + .items + .iter() + .find(|item| matches!(item, ThreadItem::UserMessage { .. })) + .cloned(); + let final_agent_message = turn + .items + .iter() + .rev() + .find(|item| matches!(item, ThreadItem::AgentMessage { .. })) + .cloned(); + turn.items = match (first_user_message, final_agent_message) { + (Some(user_message), Some(agent_message)) + if user_message.id() != agent_message.id() => + { + vec![user_message, agent_message] + } + (Some(user_message), _) => vec![user_message], + (None, Some(agent_message)) => vec![agent_message], + (None, None) => Vec::new(), + }; + turn.items_view = TurnItemsView::Summary; + } + TurnItemsView::Full => { + turn.items_view = TurnItemsView::Full; + } + } + } +} + +pub(crate) fn with_thread_spawn_agent_metadata( + source: SessionSource, + agent_nickname: Option, + agent_role: Option, +) -> SessionSource { + if agent_nickname.is_none() && agent_role.is_none() { + return source; + } + + match source { + SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth, + agent_path, + agent_nickname: existing_agent_nickname, + agent_role: existing_agent_role, + }) => SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth, + agent_path, + agent_nickname: agent_nickname.or(existing_agent_nickname), + agent_role: agent_role.or(existing_agent_role), + }), + _ => source, + } +} + +#[cfg(test)] +#[path = "thread_views_tests.rs"] +mod tests; diff --git a/codex-rs/app-server/src/thread_views_tests.rs b/codex-rs/app-server/src/thread_views_tests.rs new file mode 100644 index 000000000000..b266e9ac5f18 --- /dev/null +++ b/codex-rs/app-server/src/thread_views_tests.rs @@ -0,0 +1,200 @@ +use super::*; + +use chrono::DateTime; +use chrono::Utc; +use codex_app_server_protocol::ThreadSource as ApiThreadSource; +use codex_app_server_protocol::TurnStatus; +use codex_protocol::models::PermissionProfile; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::ThreadSource; +use pretty_assertions::assert_eq; +use std::path::PathBuf; +use uuid::Uuid; + +#[test] +fn source_filter_defaults_to_interactive_sources() { + for source_kinds in [None, Some(Vec::new())] { + let filter = ThreadSourceFilter::new(source_kinds); + + assert_eq!( + filter.store_sources(), + INTERACTIVE_SESSION_SOURCES.as_slice() + ); + assert!(filter.matches(&SessionSource::Cli)); + assert!(filter.matches(&SessionSource::Exec)); + } +} + +#[test] +fn source_filter_uses_store_filter_for_interactive_kinds() { + let filter = + ThreadSourceFilter::new(Some(vec![ThreadSourceKind::Cli, ThreadSourceKind::VsCode])); + + assert_eq!( + filter.store_sources(), + &[SessionSource::Cli, SessionSource::VSCode] + ); + assert!(filter.matches(&SessionSource::Cli)); + assert!(!filter.matches(&SessionSource::Exec)); +} + +#[test] +fn source_filter_distinguishes_subagent_variants() { + let parent_thread_id = + ThreadId::from_string(&Uuid::new_v4().to_string()).expect("valid thread id"); + let review = SessionSource::SubAgent(SubAgentSource::Review); + let spawn = SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }); + let review_filter = ThreadSourceFilter::new(Some(vec![ThreadSourceKind::SubAgentReview])); + + assert_eq!(review_filter.store_sources(), &[]); + assert!(review_filter.matches(&review)); + assert!(!review_filter.matches(&spawn)); +} + +#[test] +fn stored_thread_projection_applies_fallbacks() { + let created_at = DateTime::parse_from_rfc3339("2025-01-02T03:04:05Z") + .expect("valid timestamp") + .with_timezone(&Utc); + let updated_at = DateTime::parse_from_rfc3339("2025-01-02T03:05:06Z") + .expect("valid timestamp") + .with_timezone(&Utc); + let thread_id = + ThreadId::from_string("00000000-0000-0000-0000-000000000123").expect("valid thread id"); + let cwd = PathBuf::from("/tmp/project"); + let fallback_cwd = + AbsolutePathBuf::from_absolute_path(PathBuf::from("/tmp/fallback")).expect("absolute path"); + let stored_thread = StoredThread { + thread_id, + extra_config: None, + rollout_path: None, + forked_from_id: None, + parent_thread_id: None, + preview: "preview".to_string(), + name: Some("name".to_string()), + model_provider: String::new(), + model: None, + reasoning_effort: None, + created_at, + updated_at, + archived_at: None, + cwd: cwd.clone(), + cli_version: "0.0.0".to_string(), + source: SessionSource::Cli, + thread_source: Some(ThreadSource::User), + agent_nickname: None, + agent_role: None, + agent_path: None, + git_info: None, + approval_mode: AskForApproval::OnRequest, + permission_profile: PermissionProfile::read_only(), + token_usage: None, + first_user_message: Some("preview".to_string()), + history: None, + }; + + assert_eq!( + from_stored_thread(stored_thread, "fallback-provider", &fallback_cwd), + Thread { + id: thread_id.to_string(), + extra: None, + session_id: thread_id.to_string(), + forked_from_id: None, + parent_thread_id: None, + preview: "preview".to_string(), + ephemeral: false, + model_provider: "fallback-provider".to_string(), + created_at: created_at.timestamp(), + updated_at: updated_at.timestamp(), + status: ThreadStatus::NotLoaded, + path: None, + cwd: AbsolutePathBuf::from_absolute_path(cwd).expect("absolute path"), + cli_version: "0.0.0".to_string(), + agent_nickname: None, + agent_role: None, + source: codex_app_server_protocol::SessionSource::Cli, + thread_source: Some(ApiThreadSource::User), + git_info: None, + name: Some("name".to_string()), + turns: Vec::new(), + } + ); +} + +#[test] +fn loaded_thread_pagination_sorts_and_excludes_anchor() { + let first = "00000000-0000-0000-0000-000000000001".to_string(); + let second = "00000000-0000-0000-0000-000000000002".to_string(); + let third = "00000000-0000-0000-0000-000000000003".to_string(); + + let first_page = paginate_loaded_thread_ids( + vec![third.clone(), first.clone(), second.clone()], + /*cursor*/ None, + Some(2), + ) + .expect("first page"); + assert_eq!( + first_page, + ThreadLoadedListResponse { + data: vec![first, second.clone()], + next_cursor: Some(second.clone()), + } + ); + + assert_eq!( + paginate_loaded_thread_ids(vec![third.clone(), second.clone()], Some(&second), Some(2)) + .expect("second page"), + ThreadLoadedListResponse { + data: vec![third], + next_cursor: None, + } + ); +} + +#[test] +fn turn_pagination_preserves_order_across_pages() { + let turns = ["turn-1", "turn-2", "turn-3"] + .into_iter() + .map(turn) + .collect::>(); + let first_page = paginate_turns( + turns.clone(), + /*cursor*/ None, + Some(2), + SortDirection::Desc, + TurnItemsView::Full, + ) + .expect("first page"); + assert_eq!(first_page.data, vec![turn("turn-3"), turn("turn-2")]); + let next_cursor = first_page.next_cursor.expect("next cursor"); + + let second_page = paginate_turns( + turns, + Some(&next_cursor), + Some(2), + SortDirection::Desc, + TurnItemsView::Full, + ) + .expect("second page"); + assert_eq!(second_page.data, vec![turn("turn-1")]); + assert_eq!(second_page.next_cursor, None); +} + +fn turn(id: &str) -> Turn { + Turn { + id: id.to_string(), + items: Vec::new(), + items_view: TurnItemsView::Full, + status: TurnStatus::Completed, + error: None, + started_at: None, + completed_at: None, + duration_ms: None, + } +} From 7a321198d7af1767d0021731a55cdade3294336f Mon Sep 17 00:00:00 2001 From: Adam Van Ymeren Date: Mon, 15 Jun 2026 21:46:24 -0700 Subject: [PATCH 2/2] [thread-store] clarify empty source filters --- codex-rs/app-server/src/thread_views_tests.rs | 2 ++ codex-rs/thread-store/src/types.rs | 4 ++-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/codex-rs/app-server/src/thread_views_tests.rs b/codex-rs/app-server/src/thread_views_tests.rs index b266e9ac5f18..163a244fa733 100644 --- a/codex-rs/app-server/src/thread_views_tests.rs +++ b/codex-rs/app-server/src/thread_views_tests.rs @@ -83,6 +83,7 @@ fn stored_thread_projection_applies_fallbacks() { reasoning_effort: None, created_at, updated_at, + recency_at: updated_at, archived_at: None, cwd: cwd.clone(), cli_version: "0.0.0".to_string(), @@ -112,6 +113,7 @@ fn stored_thread_projection_applies_fallbacks() { model_provider: "fallback-provider".to_string(), created_at: created_at.timestamp(), updated_at: updated_at.timestamp(), + recency_at: Some(updated_at.timestamp()), status: ThreadStatus::NotLoaded, path: None, cwd: AbsolutePathBuf::from_absolute_path(cwd).expect("absolute path"), diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 1f761cea938c..8559906567f9 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -202,7 +202,7 @@ pub struct ListThreadsParams { pub sort_key: ThreadSortKey, /// Sort direction requested by the caller. pub sort_direction: SortDirection, - /// Allowed session sources. Empty means implementation default. + /// Allowed session sources. Empty means no source filter. pub allowed_sources: Vec, /// Optional model provider filter. `None` means implementation default, while an empty vector /// means all providers. @@ -231,7 +231,7 @@ pub struct SearchThreadsParams { pub sort_key: ThreadSortKey, /// Sort direction requested by the caller. pub sort_direction: SortDirection, - /// Allowed session sources. Empty means implementation default. + /// Allowed session sources. Empty means no source filter. pub allowed_sources: Vec, /// Whether archived threads should be searched instead of active threads. pub archived: bool,