[app-server] feat: add filtering on thread list (#9897)

codex-rs/app-server/src/codex_message_processor.rs

@@ -111,6 +111,7 @@ use codex_app_server_protocol::ThreadResumeParams;
 use codex_app_server_protocol::ThreadResumeResponse;
 use codex_app_server_protocol::ThreadRollbackParams;
 use codex_app_server_protocol::ThreadSortKey;
+use codex_app_server_protocol::ThreadSourceKind;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::ThreadStartedNotification;
@@ -132,7 +133,6 @@ use codex_chatgpt::connectors;
 use codex_core::AuthManager;
 use codex_core::CodexThread;
 use codex_core::Cursor as RolloutCursor;
-use codex_core::INTERACTIVE_SESSION_SOURCES;
 use codex_core::InitialHistory;
 use codex_core::NewThread;
 use codex_core::RolloutRecorder;
@@ -209,6 +209,9 @@ use tracing::info;
 use tracing::warn;
 use uuid::Uuid;
 
+use crate::filters::compute_source_filters;
+use crate::filters::source_kind_matches;
+
 type PendingInterruptQueue = Vec<(RequestId, ApiVersion)>;
 pub(crate) type PendingInterrupts = Arc<Mutex<HashMap<ThreadId, PendingInterruptQueue>>>;
 
@@ -1845,6 +1848,7 @@ impl CodexMessageProcessor {
             limit,
             sort_key,
             model_providers,
+            source_kinds,
             archived,
         } = params;
 
@@ -1861,6 +1865,7 @@ impl CodexMessageProcessor {
                 requested_page_size,
                 cursor,
                 model_providers,
+                source_kinds,
                 core_sort_key,
                 archived.unwrap_or(false),
             )
@@ -2538,6 +2543,7 @@ impl CodexMessageProcessor {
                 requested_page_size,
                 cursor,
                 model_providers,
+                None,
                 CoreThreadSortKey::UpdatedAt,
                 false,
             )
@@ -2558,6 +2564,7 @@ impl CodexMessageProcessor {
         requested_page_size: usize,
         cursor: Option<String>,
         model_providers: Option<Vec<String>>,
+        source_kinds: Option<Vec<ThreadSourceKind>>,
         sort_key: CoreThreadSortKey,
         archived: bool,
     ) -> Result<(Vec<ConversationSummary>, Option<String>), JSONRPCErrorError> {
@@ -2587,6 +2594,8 @@ impl CodexMessageProcessor {
             None => Some(vec![self.config.model_provider_id.clone()]),
         };
         let fallback_provider = self.config.model_provider_id.clone();
+        let (allowed_sources_vec, source_kind_filter) = compute_source_filters(source_kinds);
+        let allowed_sources = allowed_sources_vec.as_slice();
 
         while remaining > 0 {
             let page_size = remaining.min(THREAD_LIST_MAX_LIMIT);
@@ -2596,7 +2605,7 @@ impl CodexMessageProcessor {
                     page_size,
                     cursor_obj.as_ref(),
                     sort_key,
-                    INTERACTIVE_SESSION_SOURCES,
+                    allowed_sources,
                     model_provider_filter.as_deref(),
                     fallback_provider.as_str(),
                 )
@@ -2612,7 +2621,7 @@ impl CodexMessageProcessor {
                     page_size,
                     cursor_obj.as_ref(),
                     sort_key,
-                    INTERACTIVE_SESSION_SOURCES,
+                    allowed_sources,
                     model_provider_filter.as_deref(),
                     fallback_provider.as_str(),
                 )
@@ -2641,6 +2650,11 @@ impl CodexMessageProcessor {
                         updated_at,
                     )
                 })
+                .filter(|summary| {
+                    source_kind_filter
+                        .as_ref()
+                        .is_none_or(|filter| source_kind_matches(&summary.source, filter))
+                })
                 .collect::<Vec<_>>();
             if filtered.len() > remaining {
                 filtered.truncate(remaining);

codex-rs/app-server/src/filters.rs

@@ -0,0 +1,155 @@
+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<ThreadSourceKind>>,
+) -> (Vec<CoreSessionSource>, Option<Vec<ThreadSourceKind>>) {
+    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::<Vec<_>>();
+        (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(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,
+        });
+
+        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]
+        ));
+    }
+}

codex-rs/app-server/src/lib.rs

@@ -42,6 +42,7 @@ mod codex_message_processor;
 mod config_api;
 mod dynamic_tools;
 mod error_code;
+mod filters;
 mod fuzzy_file_search;
 mod message_processor;
 mod models;