diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 534886e0bf..ab47becf11 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -670,6 +670,7 @@ fn sample_thread(thread_id: &str) -> Thread { extra: None, session_id: format!("session-{thread_id}"), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: "first prompt".to_string(), ephemeral: false, diff --git a/codex-rs/analytics/src/tests/support.rs b/codex-rs/analytics/src/tests/support.rs index b1cc0dd4a9..04ba40cfc9 100644 --- a/codex-rs/analytics/src/tests/support.rs +++ b/codex-rs/analytics/src/tests/support.rs @@ -112,6 +112,7 @@ pub(super) fn sample_thread_with_metadata( extra: None, session_id: format!("session-{thread_id}"), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id, preview: "first prompt".to_string(), ephemeral, diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 561f345bbb..9295fb23f4 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index d61e1b3c72..b1e0d12b22 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -3347,6 +3347,7 @@ mod tests { extra: None, session_id: "67e55044-10b1-426f-9247-bb680e5fe0c7".to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: "first prompt".to_string(), ephemeral: true, @@ -3406,6 +3407,7 @@ mod tests { "extra": null, "sessionId": "67e55044-10b1-426f-9247-bb680e5fe0c7", "forkedFromId": null, + "attachedToThreadId": null, "parentThreadId": null, "preview": "first prompt", "ephemeral": true, diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index 238a645fc0..4a5da61b82 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -304,6 +304,7 @@ fn thread_resume_response_round_trips_initial_turns_page() { extra: None, session_id: "thr_123".to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: String::new(), ephemeral: false, diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index a5845056db..84b61dfb20 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -1435,6 +1435,10 @@ pub struct ThreadSectionDeleteResponse {} #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct ThreadListParams { + /// Include persistent threads attached to a parent when searching by title. + #[experimental("thread/list.includeAttachedThreads")] + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub include_attached_threads: bool, /// Show local history visible to the current ChatGPT account. Defaults to false. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub current_account_only: bool, @@ -1526,6 +1530,9 @@ pub struct ThreadListParams { #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct ThreadSearchParams { + /// Include persistent threads attached to a parent in search results. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub include_attached_threads: bool, /// Show local history visible to the current ChatGPT account. Defaults to false. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub current_account_only: bool, diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs index dd7c0d854a..24c497ca7b 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs @@ -218,6 +218,9 @@ pub struct Thread { pub session_id: String, /// Source thread id when this thread was created by forking another thread. pub forked_from_id: Option, + /// Parent that owns this persisted attached thread, independent of fork or spawn lineage. + #[experimental("thread.attachedToThreadId")] + pub attached_to_thread_id: Option, /// The ID of the parent thread. This will only be set if this thread is a subagent. pub parent_thread_id: Option, /// Usually the first user message in the thread, if available. @@ -305,6 +308,7 @@ struct ThreadCompatibility { extra: Option, session_id: String, forked_from_id: Option, + attached_to_thread_id: Option, parent_thread_id: Option, preview: String, ephemeral: bool, @@ -350,6 +354,7 @@ impl<'de> Deserialize<'de> for Thread { extra: thread.extra, session_id: thread.session_id, forked_from_id: thread.forked_from_id, + attached_to_thread_id: thread.attached_to_thread_id, parent_thread_id: thread.parent_thread_id, preview: thread.preview, ephemeral: thread.ephemeral, diff --git a/codex-rs/app-server-test-client/src/lib.rs b/codex-rs/app-server-test-client/src/lib.rs index c57f9758fe..97cb1fe5ec 100644 --- a/codex-rs/app-server-test-client/src/lib.rs +++ b/codex-rs/app-server-test-client/src/lib.rs @@ -1167,6 +1167,7 @@ async fn thread_list(endpoint: &Endpoint, config_overrides: &[String], limit: u3 println!("< initialize response: {initialize:?}"); let response = client.thread_list(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, 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 b4d27cde9a..b21b92ca2f 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2754,6 +2754,7 @@ impl ThreadRequestProcessor { params: ThreadListParams, ) -> Result { let ThreadListParams { + include_attached_threads, current_account_only, cursor, limit, @@ -2815,6 +2816,15 @@ impl ThreadRequestProcessor { } } let cwd_filters = normalize_thread_list_cwd_filters(cwd)?; + if include_attached_threads + && search_term + .as_ref() + .is_none_or(|term| term.trim().is_empty()) + { + return Err(invalid_request( + "includeAttachedThreads requires a non-empty searchTerm", + )); + } let relation_filter = match (parent_thread_id, ancestor_thread_id, attached_to_thread_id) { (Some(_), Some(_), None) => { return Err(invalid_request( @@ -2850,6 +2860,9 @@ impl ThreadRequestProcessor { "attachedToThreadId cannot be combined with parentThreadId or ancestorThreadId", )); } + (None, None, None) if include_attached_threads => { + Some(StoreThreadRelationFilter::IncludeAttached) + } (None, None, None) => None, }; @@ -2926,6 +2939,7 @@ impl ThreadRequestProcessor { params: ThreadSearchParams, ) -> Result { let ThreadSearchParams { + include_attached_threads, current_account_only, cursor, limit, @@ -2961,6 +2975,7 @@ impl ThreadRequestProcessor { let page = self .thread_store .search_threads(StoreSearchThreadsParams { + include_attached_threads, account: account.clone(), page_size: remaining.min(THREAD_LIST_MAX_LIMIT), cursor: cursor_obj.clone(), @@ -6078,6 +6093,8 @@ impl ThreadRequestProcessor { let mut items = Vec::with_capacity(requested_page_size); let mut next_cursor: Option = None; + let parent_scoped = + relation_filter.is_some_and(StoreThreadRelationFilter::is_parent_scoped); let model_provider_filter = match model_providers { Some(providers) => { if providers.is_empty() { @@ -6086,15 +6103,14 @@ impl ThreadRequestProcessor { Some(providers) } } - None if relation_filter.is_some() => None, + None if parent_scoped => 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_vec, source_kind_filter) = if parent_scoped && source_kinds.is_none() { + (Vec::new(), None) + } else { + compute_source_filters(source_kinds) + }; let allowed_sources = allowed_sources_vec.as_slice(); let store_sort_direction = match sort_direction { SortDirection::Asc => StoreSortDirection::Asc, @@ -6752,6 +6768,7 @@ fn thread_from_stored_thread( extra: None, session_id: thread_id, forked_from_id: thread.forked_from_id.map(|id| id.to_string()), + attached_to_thread_id: thread.attached_to_thread_id.map(|id| id.to_string()), parent_thread_id: thread.parent_thread_id.map(|id| id.to_string()), preview: thread.preview, ephemeral: false, @@ -6952,6 +6969,7 @@ fn build_thread_from_snapshot( extra: None, session_id, forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: config_snapshot.parent_thread_id.map(|id| id.to_string()), preview: String::new(), ephemeral: config_snapshot.ephemeral, 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 af0f34c561..2fc26a2e4b 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 @@ -509,6 +509,7 @@ mod thread_processor_behavior_tests { extra_config: None, rollout_path: Some(PathBuf::from("/tmp/thread.jsonl")), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: "preview".to_string(), name: None, diff --git a/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs b/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs index e33110abf9..8ae17d9a13 100644 --- a/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs +++ b/codex-rs/app-server/src/request_processors/thread_resume_redaction.rs @@ -220,6 +220,7 @@ mod tests { extra: None, session_id: "session-1".to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: "preview".to_string(), ephemeral: false, 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 42562e4084..b88c911829 100644 --- a/codex-rs/app-server/src/request_processors/thread_summary.rs +++ b/codex-rs/app-server/src/request_processors/thread_summary.rs @@ -266,6 +266,7 @@ pub(crate) fn summary_to_thread( extra: None, session_id: thread_id, forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview, ephemeral: false, diff --git a/codex-rs/app-server/tests/suite/v2/attached_thread_tests.rs b/codex-rs/app-server/tests/suite/v2/attached_thread_tests.rs index 08c95a748b..dba3700702 100644 --- a/codex-rs/app-server/tests/suite/v2/attached_thread_tests.rs +++ b/codex-rs/app-server/tests/suite/v2/attached_thread_tests.rs @@ -15,6 +15,7 @@ use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadLoadedListResponse; use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadSearchResponse; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::UserInput; @@ -75,6 +76,7 @@ async fn saved_attached_thread_upgrade_restart_and_delete(recovery: Recovery) -> let fork: ThreadForkResponse = mcp.read_response(id).await?; assert!(!fork.thread.ephemeral); assert_eq!(fork.thread.forked_from_id, Some(parent.clone())); + assert_eq!(fork.thread.attached_to_thread_id.as_ref(), Some(&parent)); children.push(fork.thread.id); } // Newer attached threads fill the raw scan page. Listing must refill across hidden @@ -162,6 +164,11 @@ async fn saved_attached_thread_upgrade_restart_and_delete(recovery: Recovery) -> found.sort(); children.sort(); assert_eq!(found, children); + let id = mcp + .send_raw_request("thread/read", Some(json!({"threadId": children[0]}))) + .await?; + let cold: ThreadReadResponse = mcp.read_response(id).await?; + assert_eq!(cold.thread.attached_to_thread_id.as_ref(), Some(&parent)); let id = mcp .send_raw_request("thread/list", Some(json!({"attachedToThreadId": other}))) .await?; @@ -187,6 +194,7 @@ async fn saved_attached_thread_upgrade_restart_and_delete(recovery: Recovery) -> }) .await?; let independent: ThreadForkResponse = mcp.read_response(id).await?; + assert_eq!(independent.thread.attached_to_thread_id, None); let id = mcp .send_raw_request( "thread/list", @@ -205,7 +213,7 @@ async fn saved_attached_thread_upgrade_restart_and_delete(recovery: Recovery) -> .send_turn_start_request(TurnStartParams { thread_id: children[0].clone(), input: vec![UserInput::Text { - text: "continue".into(), + text: "continue the chartreuse calculation".into(), text_elements: vec![], }], ..Default::default() @@ -236,6 +244,42 @@ async fn saved_attached_thread_upgrade_restart_and_delete(recovery: Recovery) -> ) .await?; let read: ThreadReadResponse = mcp.read_response(id).await?; + let id = mcp + .send_raw_request( + "thread/name/set", + Some(json!({ + "threadId": children[0], "name": "chartreuse calculation" + })), + ) + .await?; + let _: serde_json::Value = mcp.read_response(id).await?; + for include in [false, true] { + let params = + json!({"searchTerm": "chartreuse calculation", "includeAttachedThreads": include}); + let id = mcp + .send_raw_request("thread/list", Some(params.clone())) + .await?; + let by_title: ThreadListResponse = mcp.read_response(id).await?; + let id = mcp.send_raw_request("thread/search", Some(params)).await?; + let by_content: ThreadSearchResponse = mcp.read_response(id).await?; + for results in [ + by_title.data, + by_content.data.into_iter().map(|hit| hit.thread).collect(), + ] { + let destinations = results + .into_iter() + .map(|thread| (thread.id, thread.attached_to_thread_id)) + .collect::>(); + assert_eq!( + destinations, + if include { + vec![(children[0].clone(), Some(parent.clone()))] + } else { + vec![] + } + ); + } + } assert_eq!(read.thread.turns.len(), 1); assert_eq!(read.thread.turns[0].id, started.turn.id); let id = mcp diff --git a/codex-rs/app-server/tests/suite/v2/external_agent_config.rs b/codex-rs/app-server/tests/suite/v2/external_agent_config.rs index 7546304013..15a45c7567 100644 --- a/codex-rs/app-server/tests/suite/v2/external_agent_config.rs +++ b/codex-rs/app-server/tests/suite/v2/external_agent_config.rs @@ -610,6 +610,7 @@ source = {:?} let request_id = mcp .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -2067,6 +2068,7 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { let request_id = mcp .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -2262,6 +2264,7 @@ required = true let request_id = mcp .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -2351,6 +2354,7 @@ async fn external_agent_config_import_accepts_detected_session_payload_after_res let request_id = mcp .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -2437,6 +2441,7 @@ async fn external_agent_config_import_skips_already_imported_session_versions() let request_id = mcp .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -2569,6 +2574,7 @@ async fn external_agent_config_import_returns_before_background_session_import_f let request_id = mcp .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -2689,6 +2695,7 @@ async fn external_agent_config_import_compacts_huge_session_before_first_follow_ let request_id = mcp .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/app-server/tests/suite/v2/projects.rs b/codex-rs/app-server/tests/suite/v2/projects.rs index 918368504e..fb89279ea9 100644 --- a/codex-rs/app-server/tests/suite/v2/projects.rs +++ b/codex-rs/app-server/tests/suite/v2/projects.rs @@ -100,6 +100,7 @@ async fn projects_list_by_recency_and_preserve_metadata_timestamps() -> Result<( .request(|request_id| ClientRequest::ThreadList { request_id, params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -327,6 +328,7 @@ async fn projects_persist_and_assign_threads() -> Result<()> { .request(|request_id| ClientRequest::ThreadList { request_id, params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -377,6 +379,7 @@ async fn projects_persist_and_assign_threads() -> Result<()> { .request(|request_id| ClientRequest::ThreadList { request_id, params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -430,6 +433,7 @@ async fn projects_persist_and_assign_threads() -> Result<()> { .request(|request_id| ClientRequest::ThreadList { request_id, params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -628,6 +632,7 @@ async fn projects_persist_and_assign_threads() -> Result<()> { .request(|request_id| ClientRequest::ThreadList { request_id, params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -722,6 +727,7 @@ async fn deleted_project_is_dropped_before_first_durable_thread_persistence() -> .request(|request_id| ClientRequest::ThreadList { request_id, params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -915,6 +921,7 @@ async fn projects_validate_filters_cursors_and_sqlite_less_assignment() -> Resul for project_id in [String::new(), Uuid::now_v7().to_string()] { let request_id = server .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -1004,6 +1011,7 @@ async fn projects_validate_filters_cursors_and_sqlite_less_assignment() -> Resul assert_eq!(error.error.code, -32601); let list_id = unsupported_projects .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs index b59ac6e0c0..f4f72c5e1a 100644 --- a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs +++ b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs @@ -255,6 +255,7 @@ async fn thread_list_repeated_cursor_returns_internal_error() -> Result<()> { .request(ClientRequest::ThreadList { request_id: RequestId::Integer(1), params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -315,6 +316,7 @@ async fn thread_list_repeated_cursor_returns_internal_error() -> Result<()> { .request(ClientRequest::ThreadList { request_id: RequestId::Integer(2), params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -499,6 +501,7 @@ async fn thread_delete_with_non_local_thread_store_does_not_create_local_persist .request(ClientRequest::ThreadList { request_id: RequestId::Integer(3), params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs b/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs index c16a2a7928..a59eff0be6 100644 --- a/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs +++ b/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs @@ -235,6 +235,7 @@ async fn state_recovery_restores_saved_threads() -> Result<()> { .await?; let request = server .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, 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 e732b9296f..43528e6850 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -116,6 +116,7 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs async fn list_threads(mcp: &mut TestAppServer) -> Result { let list_id = mcp .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_list.rs b/codex-rs/app-server/tests/suite/v2/thread_list.rs index 1310e5af54..66a5052718 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_list.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_list.rs @@ -102,6 +102,7 @@ async fn list_threads_with_sort( mcp.request(|request_id| ClientRequest::ThreadList { request_id, params: codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -145,6 +146,7 @@ async fn list_threads_for_relation( mcp.request(|request_id| ClientRequest::ThreadList { request_id, params: codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -779,6 +781,7 @@ async fn thread_list_respects_cwd_filters() -> Result<()> { let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -822,17 +825,10 @@ async fn thread_list_respects_cwd_filters() -> Result<()> { #[tokio::test] async fn thread_list_respects_search_term_filter() -> Result<()> { let codex_home = TempDir::new()?; - std::fs::write( - codex_home.path().join("config.toml"), - r#" -model = "mock-model" -approval_policy = "never" -suppress_unstable_features_warning = true - -[features] -sqlite = true -"#, - )?; + let server = create_mock_responses_server_repeating_assistant("Done").await; + MockResponsesConfig::new(&server.uri()) + .enable_feature(Feature::Sqlite) + .write(codex_home.path())?; let older_match = create_fake_rollout( codex_home.path(), @@ -859,6 +855,24 @@ sqlite = true /*git_info*/ None, )?; + create_fake_rollout( + codex_home.path(), + "2025-01-02T13-00-00", + "2025-01-02T13:00:00Z", + "needle on another provider", + Some("other_provider"), + /*git_info*/ None, + )?; + let exec_match = create_fake_rollout_with_source( + codex_home.path(), + "2025-01-02T14-00-00", + "2025-01-02T14:00:00Z", + "exec needle", + Some("mock_provider"), + /*git_info*/ None, + CoreSessionSource::Exec, + )?; + // `thread/list` applies `search_term` on the sqlite fast path. This test creates // rollouts manually, so mark the DB backfill complete and then run an unsearched // list large enough to repair every rollout the searched list should find. @@ -889,40 +903,32 @@ sqlite = true /*cwd_filters*/ None, "mock_provider", /*search_term*/ None, + /*include_attached_threads*/ false, ) .await?; - assert_eq!(repaired_page.items.len(), 3); + assert_eq!(repaired_page.items.len(), 5); let mut mcp = init_mcp(codex_home.path()).await?; - let request_id = mcp - .send_thread_list_request(codex_app_server_protocol::ThreadListParams { - current_account_only: false, - excluded_thread_ids: None, - attached_to_thread_id: None, - originators: None, - cursor: None, - limit: Some(10), - sort_key: None, - sort_direction: None, - model_providers: Some(vec!["mock_provider".to_string()]), - source_kinds: None, - archived: None, - section_id: None, - project_id: None, - cwd: None, - use_state_db_only: false, - search_term: Some("needle".to_string()), - parent_thread_id: None, - ancestor_thread_id: None, - }) - .await?; - let ThreadListResponse { - data, next_cursor, .. - } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(request_id)).await??; - - assert_eq!(next_cursor, None); - let ids: Vec<_> = data.iter().map(|thread| thread.id.as_str()).collect(); - assert_eq!(ids, vec![newer_match, older_match]); + for (mut params, expected_ids) in [ + (json!({}), vec![&newer_match, &older_match]), + ( + json!({"includeAttachedThreads": true}), + vec![&newer_match, &older_match], + ), + ( + json!({"includeAttachedThreads": true, "sourceKinds": ["exec"]}), + vec![&exec_match], + ), + ] { + params["limit"] = json!(10); + params["searchTerm"] = json!("needle"); + let request_id = mcp.send_raw_request("thread/list", Some(params)).await?; + let response: ThreadListResponse = + timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(request_id)).await??; + assert_eq!(response.next_cursor, None); + let ids: Vec<_> = response.data.iter().map(|thread| &thread.id).collect(); + assert_eq!(ids, expected_ids); + } Ok(()) } @@ -968,6 +974,7 @@ async fn thread_search_returns_content_matches() -> Result<()> { let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_search_request(codex_app_server_protocol::ThreadSearchParams { + include_attached_threads: false, current_account_only: false, cursor: None, limit: Some(10), @@ -1024,6 +1031,7 @@ async fn thread_search_returns_content_matches() -> Result<()> { let request_id = mcp .send_thread_search_request(codex_app_server_protocol::ThreadSearchParams { + include_attached_threads: false, current_account_only: false, cursor: None, limit: Some(10), @@ -1087,6 +1095,7 @@ async fn thread_search_matches_json_escaped_content() -> Result<()> { let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_search_request(codex_app_server_protocol::ThreadSearchParams { + include_attached_threads: false, current_account_only: false, cursor: None, limit: Some(10), @@ -1133,6 +1142,7 @@ async fn thread_search_filters_by_source_kind() -> Result<()> { let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_search_request(codex_app_server_protocol::ThreadSearchParams { + include_attached_threads: false, current_account_only: false, cursor: None, limit: Some(10), @@ -1311,6 +1321,7 @@ sqlite = true let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -1351,6 +1362,7 @@ sqlite = true let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -1384,6 +1396,7 @@ sqlite = true let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -1573,6 +1586,7 @@ async fn thread_list_relation_filters_reject_invalid_requests() -> Result<()> { let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -1603,6 +1617,7 @@ async fn thread_list_relation_filters_reject_invalid_requests() -> Result<()> { let thread_id = ThreadId::new().to_string(); let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -1844,6 +1859,7 @@ async fn thread_list_reports_loaded_subagent_direct_input_capability() -> Result let request_id = mcp .send_thread_search_request(codex_app_server_protocol::ThreadSearchParams { + include_attached_threads: false, current_account_only: false, cursor: None, limit: Some(10), @@ -1898,6 +1914,7 @@ async fn thread_list_reports_loaded_subagent_direct_input_capability() -> Result .request(|request_id| ClientRequest::ThreadList { request_id, params: codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -2567,6 +2584,7 @@ async fn thread_list_sort_recency_at_uses_state_db_order_with_provider_filter() /*cwd_filters*/ None, "mock_provider", /*search_term*/ None, + /*include_attached_threads*/ false, ) .await?; state_db @@ -2723,6 +2741,7 @@ async fn thread_list_backwards_cursor_can_seed_forward_delta_sync() -> Result<() } = { let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -2768,6 +2787,7 @@ async fn thread_list_backwards_cursor_can_seed_forward_delta_sync() -> Result<() } = { let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -3203,6 +3223,7 @@ async fn thread_list_invalid_cursor_returns_error() -> Result<()> { let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_metadata_update.rs b/codex-rs/app-server/tests/suite/v2/thread_metadata_update.rs index 17d053bdd3..04a7479158 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_metadata_update.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_metadata_update.rs @@ -88,6 +88,7 @@ async fn thread_section_move_pins_and_archives_before_first_turn() -> Result<()> timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(move_id)).await??; let list_params = ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -295,6 +296,7 @@ async fn thread_section_move_pins_and_unpins_with_filtered_recency_pagination() } let list_params = ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -490,6 +492,7 @@ async fn thread_sections_preserve_server_owned_manual_order_across_moves_and_res } let list_params = ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index f7ad67d679..6cd281857f 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -415,6 +415,7 @@ async fn paginated_stored_thread_routes_projected_turns() -> Result<()> { let list_id = mcp .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -1161,6 +1162,7 @@ async fn thread_list_includes_store_thread_without_rollout_path() -> Result<()> .request(ClientRequest::ThreadList { request_id: RequestId::Integer(1), params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, @@ -1644,6 +1646,7 @@ async fn paginated_thread_name_set_is_reflected_in_read_list_and_metadata_resume // List should also surface the name. let list_id = mcp .send_thread_list_request(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 34e9d2dc45..44014ab340 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -4722,6 +4722,7 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected( .request(|request_id| ClientRequest::ThreadList { request_id, params: codex_app_server_protocol::ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/cli/tests/queue.rs b/codex-rs/cli/tests/queue.rs index 23c241ff58..13f2554b98 100644 --- a/codex-rs/cli/tests/queue.rs +++ b/codex-rs/cli/tests/queue.rs @@ -474,6 +474,7 @@ async fn remote_session_commands_with_workload_identity_use_server_auth() -> Res .request_typed(ClientRequest::ThreadList { request_id: RequestId::Integer(4), params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/core/src/realtime_context_tests.rs b/codex-rs/core/src/realtime_context_tests.rs index 2faf5c0649..2f7da06c59 100644 --- a/codex-rs/core/src/realtime_context_tests.rs +++ b/codex-rs/core/src/realtime_context_tests.rs @@ -36,6 +36,7 @@ fn stored_thread(cwd: &str, title: &str, first_user_message: &str) -> StoredThre extra_config: None, rollout_path: Some(PathBuf::from("/tmp/rollout.jsonl")), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: first_user_message.to_string(), name: (!title.is_empty()).then(|| title.to_string()), diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 827eb2be00..bb01fc3568 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -2007,6 +2007,7 @@ async fn resolve_resume_thread_id( ClientRequest::ThreadList { request_id: RequestId::Integer(0), params: ThreadListParams { + include_attached_threads: false, current_account_only: true, excluded_thread_ids: None, attached_to_thread_id: None, @@ -2096,6 +2097,7 @@ async fn resolve_resume_thread_id( ClientRequest::ThreadList { request_id: RequestId::Integer(0), params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/exec/src/lib_tests.rs b/codex-rs/exec/src/lib_tests.rs index af43049b02..b518d37265 100644 --- a/codex-rs/exec/src/lib_tests.rs +++ b/codex-rs/exec/src/lib_tests.rs @@ -358,6 +358,7 @@ fn turn_items_for_thread_returns_matching_turn_items() { extra: None, session_id: "thread-1".to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: String::new(), ephemeral: false, @@ -871,6 +872,7 @@ fn sample_thread_start_response() -> ThreadStartResponse { extra: None, session_id: "67e55044-10b1-426f-9247-bb680e5fe0c7".to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: String::new(), ephemeral: false, diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index ec40f50ea6..60b9972cc0 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -88,6 +88,8 @@ pub struct ThreadItem { pub source: Option, /// Persisted thread history contract selected when this thread was created. pub history_mode: ThreadHistoryMode, + /// Parent of a persisted attached thread. + pub attached_to_thread_id: Option, /// Immediate control/spawn parent thread id from session metadata. pub parent_thread_id: Option, /// Random unique nickname from session metadata for AgentControl-spawned sub-agents. @@ -123,7 +125,7 @@ struct HeadTailSummary { originator: Option, saw_session_meta: bool, has_fork_history: bool, - attached_to_parent: bool, + attached_to_thread_id: Option, thread_id: Option, first_user_message: Option, preview: Option, @@ -166,12 +168,14 @@ pub enum ThreadListLayout { Flat, } +#[derive(Clone, Copy)] pub struct ThreadListConfig<'a> { pub allowed_sources: &'a [SessionSource], pub model_providers: Option<&'a [String]>, pub cwd_filters: Option<&'a [PathBuf]>, pub default_provider: &'a str, pub layout: ThreadListLayout, + pub include_attached_threads: bool, } /// Pagination cursor identifying the last item in a page. @@ -256,9 +260,8 @@ struct FilesByCreatedAtVisitor<'a> { page_size: usize, anchor_state: AnchorState, more_matches_available: bool, - allowed_sources: &'a [SessionSource], + config: ThreadListConfig<'a>, provider_matcher: Option<&'a ProviderMatcher<'a>>, - cwd_filters: Option<&'a [PathBuf]>, } impl<'a> RolloutFileVisitor for FilesByCreatedAtVisitor<'a> { @@ -286,12 +289,12 @@ impl<'a> RolloutFileVisitor for FilesByCreatedAtVisitor<'a> { .and_then(format_rfc3339); if let Some(item) = build_thread_item( path, - self.allowed_sources, + self.config.allowed_sources, self.provider_matcher, - self.cwd_filters, + self.config.cwd_filters, updated_at, PreviewFilter::RequirePreview, - /*include_attached_threads*/ false, + self.config.include_attached_threads, ) .await { @@ -364,6 +367,7 @@ pub async fn get_threads( cwd_filters, default_provider, layout: ThreadListLayout::NestedByDate, + include_attached_threads: false, }, ) .await @@ -398,9 +402,8 @@ pub async fn get_threads_in_root( page_size, anchor, sort_key, - config.allowed_sources, + config, provider_matcher.as_ref(), - config.cwd_filters, ) .await? } @@ -410,9 +413,8 @@ pub async fn get_threads_in_root( page_size, anchor, sort_key, - config.allowed_sources, + config, provider_matcher.as_ref(), - config.cwd_filters, ) .await? } @@ -429,9 +431,8 @@ async fn traverse_directories_for_paths( page_size: usize, anchor: Option, sort_key: ThreadSortKey, - allowed_sources: &[SessionSource], + config: ThreadListConfig<'_>, provider_matcher: Option<&ProviderMatcher<'_>>, - cwd_filters: Option<&[PathBuf]>, ) -> io::Result { match sort_key { ThreadSortKey::CreatedAt => { @@ -439,9 +440,8 @@ async fn traverse_directories_for_paths( root, page_size, anchor, - allowed_sources, + config, provider_matcher, - cwd_filters, ) .await } @@ -450,9 +450,8 @@ async fn traverse_directories_for_paths( root, page_size, anchor, - allowed_sources, + config, provider_matcher, - cwd_filters, ) .await } @@ -464,32 +463,15 @@ async fn traverse_flat_paths( page_size: usize, anchor: Option, sort_key: ThreadSortKey, - allowed_sources: &[SessionSource], + config: ThreadListConfig<'_>, provider_matcher: Option<&ProviderMatcher<'_>>, - cwd_filters: Option<&[PathBuf]>, ) -> io::Result { match sort_key { ThreadSortKey::CreatedAt => { - traverse_flat_paths_created( - root, - page_size, - anchor, - allowed_sources, - provider_matcher, - cwd_filters, - ) - .await + traverse_flat_paths_created(root, page_size, anchor, config, provider_matcher).await } ThreadSortKey::UpdatedAt | ThreadSortKey::RecencyAt => { - traverse_flat_paths_updated( - root, - page_size, - anchor, - allowed_sources, - provider_matcher, - cwd_filters, - ) - .await + traverse_flat_paths_updated(root, page_size, anchor, config, provider_matcher).await } } } @@ -504,9 +486,8 @@ async fn traverse_directories_for_paths_created( root: PathBuf, page_size: usize, anchor: Option, - allowed_sources: &[SessionSource], + config: ThreadListConfig<'_>, provider_matcher: Option<&ProviderMatcher<'_>>, - cwd_filters: Option<&[PathBuf]>, ) -> io::Result { let mut items: Vec = Vec::with_capacity(page_size); let mut scanned_files = 0usize; @@ -516,9 +497,8 @@ async fn traverse_directories_for_paths_created( page_size, anchor_state: AnchorState::new(anchor), more_matches_available, - allowed_sources, + config, provider_matcher, - cwd_filters, }; walk_rollout_files(&root, &mut scanned_files, &mut visitor).await?; more_matches_available = visitor.more_matches_available; @@ -555,9 +535,8 @@ async fn traverse_directories_for_paths_updated( root: PathBuf, page_size: usize, anchor: Option, - allowed_sources: &[SessionSource], + config: ThreadListConfig<'_>, provider_matcher: Option<&ProviderMatcher<'_>>, - cwd_filters: Option<&[PathBuf]>, ) -> io::Result { let mut items: Vec = Vec::with_capacity(page_size); let mut scanned_files = 0usize; @@ -583,12 +562,12 @@ async fn traverse_directories_for_paths_updated( let updated_at_fallback = format_rfc3339(ts); if let Some(item) = build_thread_item( candidate.path, - allowed_sources, + config.allowed_sources, provider_matcher, - cwd_filters, + config.cwd_filters, updated_at_fallback, PreviewFilter::RequirePreview, - /*include_attached_threads*/ false, + config.include_attached_threads, ) .await { @@ -620,9 +599,8 @@ async fn traverse_flat_paths_created( root: PathBuf, page_size: usize, anchor: Option, - allowed_sources: &[SessionSource], + config: ThreadListConfig<'_>, provider_matcher: Option<&ProviderMatcher<'_>>, - cwd_filters: Option<&[PathBuf]>, ) -> io::Result { let mut items: Vec = Vec::with_capacity(page_size); let mut scanned_files = 0usize; @@ -644,12 +622,12 @@ async fn traverse_flat_paths_created( .and_then(format_rfc3339); if let Some(item) = build_thread_item( path, - allowed_sources, + config.allowed_sources, provider_matcher, - cwd_filters, + config.cwd_filters, updated_at, PreviewFilter::IncludeEmpty, - /*include_attached_threads*/ false, + config.include_attached_threads, ) .await { @@ -681,9 +659,8 @@ async fn traverse_flat_paths_updated( root: PathBuf, page_size: usize, anchor: Option, - allowed_sources: &[SessionSource], + config: ThreadListConfig<'_>, provider_matcher: Option<&ProviderMatcher<'_>>, - cwd_filters: Option<&[PathBuf]>, ) -> io::Result { let mut items: Vec = Vec::with_capacity(page_size); let mut scanned_files = 0usize; @@ -709,12 +686,12 @@ async fn traverse_flat_paths_updated( let updated_at_fallback = format_rfc3339(ts); if let Some(item) = build_thread_item( candidate.path, - allowed_sources, + config.allowed_sources, provider_matcher, - cwd_filters, + config.cwd_filters, updated_at_fallback, PreviewFilter::IncludeEmpty, - /*include_attached_threads*/ false, + config.include_attached_threads, ) .await { @@ -816,7 +793,7 @@ async fn build_thread_item( let summary = read_head_summary(&path, HEAD_RECORD_LIMIT) .await .unwrap_or_default(); - if summary.attached_to_parent && !include_attached_threads { + if summary.attached_to_thread_id.is_some() && !include_attached_threads { return None; } if !allowed_sources.is_empty() @@ -844,7 +821,8 @@ async fn build_thread_item( if summary.saw_session_meta && (matches!(preview_filter, PreviewFilter::IncludeEmpty) || summary.preview.is_some() - || summary.has_fork_history) + || summary.has_fork_history + || summary.attached_to_thread_id.is_some()) { let HeadTailSummary { originator, @@ -873,6 +851,7 @@ async fn build_thread_item( originator, path, thread_id, + attached_to_thread_id: summary.attached_to_thread_id, first_user_message, preview, section: None, @@ -1035,7 +1014,10 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result, default_provider: &str, search_term: Option<&str>, + include_attached_threads: bool, ) -> std::io::Result { Self::list_threads_with_db_fallback( state_db_ctx, @@ -431,6 +432,7 @@ impl RolloutRecorder { ThreadListArchiveFilter::Active, ThreadListRepairMode::ScanAndRepair, search_term, + include_attached_threads, ) .await } @@ -463,6 +465,7 @@ impl RolloutRecorder { ThreadListArchiveFilter::Active, ThreadListRepairMode::StateDbOnly, search_term, + /*include_attached_threads*/ false, ) .await } @@ -481,6 +484,7 @@ impl RolloutRecorder { cwd_filters: Option<&[PathBuf]>, default_provider: &str, search_term: Option<&str>, + include_attached_threads: bool, ) -> std::io::Result { Self::list_threads_with_db_fallback( state_db_ctx, @@ -496,6 +500,7 @@ impl RolloutRecorder { ThreadListArchiveFilter::Archived, ThreadListRepairMode::ScanAndRepair, search_term, + include_attached_threads, ) .await } @@ -528,6 +533,7 @@ impl RolloutRecorder { ThreadListArchiveFilter::Archived, ThreadListRepairMode::StateDbOnly, search_term, + /*include_attached_threads*/ false, ) .await } @@ -547,6 +553,7 @@ impl RolloutRecorder { archive_filter: ThreadListArchiveFilter, repair_mode: ThreadListRepairMode, search_term: Option<&str>, + include_attached_threads: bool, ) -> std::io::Result { let codex_home = config.codex_home(); let sqlite = config.sqlite_config(); @@ -554,6 +561,8 @@ impl RolloutRecorder { ThreadListArchiveFilter::Active => false, ThreadListArchiveFilter::Archived => true, }; + let relation_filter = + include_attached_threads.then_some(codex_state::ThreadRelationFilter::IncludeAttached); if matches!(repair_mode, ThreadListRepairMode::StateDbOnly) { return state_db::list_threads_db( state_db_ctx.as_deref(), @@ -565,7 +574,7 @@ impl RolloutRecorder { allowed_sources, model_providers, cwd_filters, - /*relation_filter*/ None, + relation_filter, archived, /*section*/ None, /*project_id*/ None, @@ -586,6 +595,18 @@ impl RolloutRecorder { // Filesystem-first listing intentionally overfetches so we can repair stale/missing // SQLite rows before returning the scan page for filtered listings or the DB page for // unfiltered listings. + let file_config = ThreadListConfig { + allowed_sources, + model_providers, + cwd_filters, + default_provider, + layout: if archived { + ThreadListLayout::Flat + } else { + ThreadListLayout::NestedByDate + }, + include_attached_threads, + }; let fs_page = match sort_direction { SortDirection::Asc => { list_threads_from_files_asc( @@ -593,11 +614,7 @@ impl RolloutRecorder { page_size, cursor, sort_key, - allowed_sources, - model_providers, - cwd_filters, - default_provider, - archived, + file_config, search_term, ) .await? @@ -608,18 +625,14 @@ impl RolloutRecorder { page_size.saturating_mul(2), cursor, sort_key, - allowed_sources, - model_providers, - cwd_filters, - default_provider, - archived, + file_config, search_term, ) .await? } }; - if state_db_ctx.is_none() { + let Some(state_db) = state_db_ctx.as_deref() else { // Keep legacy behavior when SQLite is unavailable: return filesystem results // at the requested page size. codex_state::record_fallback( @@ -633,7 +646,7 @@ impl RolloutRecorder { page_size, sort_key, )); - } + }; // For metadata-filtered listings the filesystem page is the page we return. Track those // IDs so the later DB page only triggers full reconciliation for DB-only hits. @@ -648,14 +661,12 @@ impl RolloutRecorder { // enough there. Search can depend on full title metadata, so keep full reconciliation. for item in &fs_page.items { if search_term.is_some() { - state_db::reconcile_rollout( - state_db_ctx.as_deref(), + state_db::reconcile_rollout_from_disk( + state_db, item.path.as_path(), default_provider, - /*builder*/ None, - &[], Some(archived), - /*new_thread_memory_mode*/ None, + Some(codex_home), ) .await; } else { @@ -679,7 +690,7 @@ impl RolloutRecorder { allowed_sources, model_providers, cwd_filters, - /*relation_filter*/ None, + relation_filter, archived, /*section*/ None, /*project_id*/ None, @@ -710,7 +721,7 @@ impl RolloutRecorder { allowed_sources, model_providers, cwd_filters, - /*relation_filter*/ None, + relation_filter, archived, /*section*/ None, /*project_id*/ None, @@ -752,7 +763,7 @@ impl RolloutRecorder { allowed_sources, model_providers, cwd_filters, - /*relation_filter*/ None, + relation_filter, archived, /*section*/ None, /*project_id*/ None, @@ -1415,6 +1426,7 @@ fn fill_missing_thread_item_metadata(item: &mut ThreadItem, state_item: ThreadIt git_origin_url, source, history_mode: _, + attached_to_thread_id, parent_thread_id, agent_nickname, agent_role, @@ -1465,6 +1477,7 @@ fn fill_missing_thread_item_metadata(item: &mut ThreadItem, state_item: ThreadIt if item.source.is_none() { item.source = source; } + item.attached_to_thread_id = item.attached_to_thread_id.or(attached_to_thread_id); if item.parent_thread_id.is_none() { item.parent_thread_id = parent_thread_id; } @@ -1491,17 +1504,12 @@ fn fill_missing_thread_item_metadata(item: &mut ThreadItem, state_item: ThreadIt } } -#[allow(clippy::too_many_arguments)] async fn list_threads_from_files_desc( codex_home: &Path, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, - allowed_sources: &[SessionSource], - model_providers: Option<&[String]>, - cwd_filters: Option<&[PathBuf]>, - default_provider: &str, - archived: bool, + config: ThreadListConfig<'_>, search_term: Option<&str>, ) -> std::io::Result { if let Some(search_term) = search_term { @@ -1517,11 +1525,7 @@ async fn list_threads_from_files_desc( scan_page_size, page_cursor.as_ref(), sort_key, - allowed_sources, - model_providers, - cwd_filters, - default_provider, - archived, + config, ) .await?; scanned_files = scanned_files.saturating_add(page.num_scanned_files); @@ -1554,74 +1558,29 @@ async fn list_threads_from_files_desc( }); } - list_threads_from_files_desc_unfiltered( - codex_home, - page_size, - cursor, - sort_key, - allowed_sources, - model_providers, - cwd_filters, - default_provider, - archived, - ) - .await + list_threads_from_files_desc_unfiltered(codex_home, page_size, cursor, sort_key, config).await } -#[allow(clippy::too_many_arguments)] async fn list_threads_from_files_desc_unfiltered( codex_home: &Path, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, - allowed_sources: &[SessionSource], - model_providers: Option<&[String]>, - cwd_filters: Option<&[PathBuf]>, - default_provider: &str, - archived: bool, + config: ThreadListConfig<'_>, ) -> std::io::Result { - if archived { - let root = codex_home.join(ARCHIVED_SESSIONS_SUBDIR); - get_threads_in_root( - root, - page_size, - cursor, - sort_key, - ThreadListConfig { - allowed_sources, - model_providers, - cwd_filters, - default_provider, - layout: ThreadListLayout::Flat, - }, - ) - .await - } else { - get_threads( - codex_home, - page_size, - cursor, - sort_key, - allowed_sources, - model_providers, - cwd_filters, - default_provider, - ) - .await - } + let root = codex_home.join(match config.layout { + ThreadListLayout::Flat => ARCHIVED_SESSIONS_SUBDIR, + ThreadListLayout::NestedByDate => SESSIONS_SUBDIR, + }); + get_threads_in_root(root, page_size, cursor, sort_key, config).await } -#[allow(clippy::too_many_arguments)] async fn list_threads_from_files_asc( codex_home: &Path, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, - allowed_sources: &[SessionSource], - model_providers: Option<&[String]>, - cwd_filters: Option<&[PathBuf]>, - default_provider: &str, - archived: bool, + config: ThreadListConfig<'_>, search_term: Option<&str>, ) -> std::io::Result { let mut all_items = Vec::new(); @@ -1635,11 +1594,7 @@ async fn list_threads_from_files_asc( scan_page_size, page_cursor.as_ref(), sort_key, - allowed_sources, - model_providers, - cwd_filters, - default_provider, - archived, + config, /*search_term*/ None, ) .await?; @@ -2165,6 +2120,7 @@ fn thread_item_from_state_metadata( .unwrap_or(SessionSource::Unknown), ), history_mode: item.history_mode, + attached_to_thread_id: item.attached_thread_parent_id, parent_thread_id, agent_nickname: item.agent_nickname, agent_role: item.agent_role, diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index c3c27f6d2e..0f88cb986f 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -1344,6 +1344,7 @@ async fn discovery_without_database_skips_attached_threads_before_pagination() - /*cwd_filters*/ None, default_provider.as_str(), /*search_term*/ None, + /*include_attached_threads*/ false, ) .await?; assert_eq!(page1.items.len(), 1); @@ -1362,6 +1363,7 @@ async fn discovery_without_database_skips_attached_threads_before_pagination() - /*cwd_filters*/ None, default_provider.as_str(), /*search_term*/ None, + /*include_attached_threads*/ false, ) .await?; assert_eq!(page2.items.len(), 1); @@ -1400,6 +1402,7 @@ async fn list_archived_threads_without_db_keeps_threads_without_previews() -> st /*cwd_filters*/ None, config.model_provider_id.as_str(), /*search_term*/ None, + /*include_attached_threads*/ false, ) .await?; assert_eq!(page.items.len(), 1); @@ -1495,6 +1498,7 @@ async fn list_threads_db_enabled_preserves_metadata_for_missing_rollout_paths() /*cwd_filters*/ None, default_provider.as_str(), /*search_term*/ None, + /*include_attached_threads*/ false, ) .await?; assert_eq!(page.items.len(), 1); @@ -1524,6 +1528,7 @@ async fn list_threads_db_enabled_preserves_metadata_for_missing_rollout_paths() /*cwd_filters*/ None, default_provider.as_str(), /*search_term*/ None, + /*include_attached_threads*/ false, ) .await?; assert_eq!(fallback.items.len(), 1); @@ -1608,6 +1613,7 @@ async fn list_threads_state_db_only_skips_jsonl_repair_scan() -> std::io::Result /*cwd_filters*/ Some(cwd_filters.as_slice()), config.model_provider_id.as_str(), /*search_term*/ None, + /*include_attached_threads*/ false, ) .await?; assert_eq!(repaired_page.items.len(), 1); @@ -1716,6 +1722,7 @@ async fn list_threads_default_filter_returns_filesystem_scan_results() -> std::i /*cwd_filters*/ Some(cwd_filters.as_slice()), config.model_provider_id.as_str(), /*search_term*/ None, + /*include_attached_threads*/ false, ) .await?; assert_eq!(scanned_page.items.len(), 0); @@ -1794,6 +1801,7 @@ async fn list_threads_metadata_filter_overlays_state_db_list_metadata() -> std:: /*cwd_filters*/ None, config.model_provider_id.as_str(), /*search_term*/ None, + /*include_attached_threads*/ false, ) .await?; @@ -1831,6 +1839,7 @@ fn fill_missing_thread_item_metadata_preserves_identity_and_prefers_state_git_fi ), source: None, history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, @@ -1864,6 +1873,7 @@ fn fill_missing_thread_item_metadata_preserves_identity_and_prefers_state_git_fi ), source: Some(SessionSource::Exec), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: Some("state-agent".to_string()), agent_role: Some("state-role".to_string()), @@ -1979,6 +1989,7 @@ async fn list_threads_search_repairs_stale_state_db_hits_before_returning() -> s /*cwd_filters*/ None, config.model_provider_id.as_str(), Some("needle"), + /*include_attached_threads*/ false, ) .await?; assert_eq!(scanned_page.items.len(), 0); diff --git a/codex-rs/rollout/src/state_db.rs b/codex-rs/rollout/src/state_db.rs index a53b099bd9..82ec525896 100644 --- a/codex-rs/rollout/src/state_db.rs +++ b/codex-rs/rollout/src/state_db.rs @@ -401,31 +401,6 @@ pub async fn list_threads_db( SortDirection::Desc => codex_state::SortDirection::Desc, }; - if let Some(relation_filter) = relation_filter { - let filters = codex_state::ThreadFilterOptions { - archived_only: archived, - section, - allowed_sources: allowed_sources.as_slice(), - model_providers: model_providers.as_deref(), - cwd_filters: normalized_cwd_filters.as_deref(), - project_id, - anchor: anchor.as_ref(), - sort_key: state_sort_key, - sort_direction: state_sort_direction, - search_term, - }; - return match ctx - .list_threads_by_relation(page_size, relation_filter, filters) - .await - { - Ok(page) => Some(page), - Err(err) => { - warn!("state db list_threads failed: {err}"); - None - } - }; - } - let mut items = Vec::with_capacity(page_size); let mut next_anchor = None; let mut num_scanned_rows = 0usize; @@ -442,13 +417,21 @@ pub async fn list_threads_db( sort_direction: state_sort_direction, search_term, }; - let page = match ctx.list_threads(page_size - items.len(), filters).await { - Ok(page) => page, - Err(err) => { - warn!("state db list_threads failed: {err}"); - return None; + let page = match relation_filter { + Some(relation) => { + ctx.list_threads_by_relation(page_size - items.len(), relation, filters) + .await } - }; + None => ctx.list_threads(page_size - items.len(), filters).await, + } + .inspect_err(|err| warn!("state db list_threads failed: {err}")) + .ok()?; + // Parent-scoped listings include empty threads whose rollout may still be pending. + if relation_filter + .is_some_and(|relation| relation != codex_state::ThreadRelationFilter::IncludeAttached) + { + return Some(page); + } num_scanned_rows = num_scanned_rows.saturating_add(page.num_scanned_rows); next_anchor = page.next_anchor; @@ -542,6 +525,23 @@ pub async fn reconcile_rollout( .await; return; } + reconcile_rollout_from_disk( + ctx, + rollout_path, + default_provider, + archived_only, + /*name_index_home*/ None, + ) + .await; +} + +pub(super) async fn reconcile_rollout_from_disk( + ctx: &codex_state::StateRuntime, + rollout_path: &Path, + default_provider: &str, + archived_only: Option, + name_index_home: Option<&Path>, +) { let outcome = match metadata::extract_metadata_from_rollout(rollout_path, default_provider).await { Ok(outcome) => outcome, @@ -573,6 +573,14 @@ pub async fn reconcile_rollout( if let Some(existing_metadata) = existing_metadata.as_ref() { metadata.prefer_existing_git_info(existing_metadata); metadata.prefer_existing_explicit_title(existing_metadata); + } else if let Some(home) = name_index_home { + // The saved name is outside the rollout. Seed it when title search repairs a missing row. + // The upsert preserves the database name if another writer creates the row first. + metadata.name = crate::find_thread_name_by_id(home, &metadata.id) + .await + .ok() + .flatten() + .filter(|name| !name.is_empty()); } match archived_only { Some(true) if metadata.archived_at.is_none() => { diff --git a/codex-rs/rollout/src/tests.rs b/codex-rs/rollout/src/tests.rs index 91be8e9a3c..f9855b313f 100644 --- a/codex-rs/rollout/src/tests.rs +++ b/codex-rs/rollout/src/tests.rs @@ -805,6 +805,7 @@ async fn test_list_conversations_latest_first() { git_origin_url: None, source: Some(SessionSource::VSCode), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, @@ -831,6 +832,7 @@ async fn test_list_conversations_latest_first() { git_origin_url: None, source: Some(SessionSource::VSCode), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, @@ -857,6 +859,7 @@ async fn test_list_conversations_latest_first() { git_origin_url: None, source: Some(SessionSource::VSCode), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, @@ -976,6 +979,7 @@ async fn test_pagination_cursor() { git_origin_url: None, source: Some(SessionSource::VSCode), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, @@ -1002,6 +1006,7 @@ async fn test_pagination_cursor() { git_origin_url: None, source: Some(SessionSource::VSCode), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, @@ -1064,6 +1069,7 @@ async fn test_pagination_cursor() { git_origin_url: None, source: Some(SessionSource::VSCode), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, @@ -1090,6 +1096,7 @@ async fn test_pagination_cursor() { git_origin_url: None, source: Some(SessionSource::VSCode), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, @@ -1144,6 +1151,7 @@ async fn test_pagination_cursor() { git_origin_url: None, source: Some(SessionSource::VSCode), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, @@ -1353,6 +1361,7 @@ async fn test_get_thread_contents() { git_origin_url: None, source: Some(SessionSource::VSCode), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, @@ -1771,6 +1780,7 @@ async fn test_filesystem_cursor_pages_through_same_timestamp_ties() { git_origin_url: None, source: Some(SessionSource::VSCode), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, @@ -1797,6 +1807,7 @@ async fn test_filesystem_cursor_pages_through_same_timestamp_ties() { git_origin_url: None, source: Some(SessionSource::VSCode), history_mode: Default::default(), + attached_to_thread_id: None, parent_thread_id: None, agent_nickname: None, agent_role: None, diff --git a/codex-rs/state/src/model/thread_metadata.rs b/codex-rs/state/src/model/thread_metadata.rs index 75378446c4..cf72159b85 100644 --- a/codex-rs/state/src/model/thread_metadata.rs +++ b/codex-rs/state/src/model/thread_metadata.rs @@ -36,6 +36,8 @@ pub enum SortDirection { /// Thread relationship used to filter thread listings. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ThreadRelationFilter { + /// Include attached threads alongside ordinary threads for search. + IncludeAttached, /// Return only threads whose immediate parent is the given thread. DirectChildrenOf(ThreadId), /// Return every thread transitively descended from the given thread. diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 85ac134aae..c398140b5a 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -1370,7 +1370,7 @@ WITH RECURSIVE subtree(child_thread_id, parent_thread_id) AS ( Some(crate::ThreadRelationFilter::DescendantsOf(_)) => builder.push( ", subtree.parent_thread_id AS parent_thread_id\nFROM subtree\nCROSS JOIN threads ON threads.id = subtree.child_thread_id", ), - Some(crate::ThreadRelationFilter::AttachedThreadsOf(_)) => { + Some(crate::ThreadRelationFilter::AttachedThreadsOf(_) | crate::ThreadRelationFilter::IncludeAttached) => { builder.push(", NULL AS parent_thread_id\nFROM threads") } None => builder.push(" FROM threads"), @@ -1379,11 +1379,14 @@ WITH RECURSIVE subtree(child_thread_id, parent_thread_id) AS ( builder, filters, /*include_thread_id_tiebreaker*/ true, - /*include_empty_preview*/ relation_filter.is_some(), + relation_filter, ); if !matches!( relation_filter, - Some(crate::ThreadRelationFilter::AttachedThreadsOf(_)) + Some( + crate::ThreadRelationFilter::AttachedThreadsOf(_) + | crate::ThreadRelationFilter::IncludeAttached + ) ) { builder.push(EXCLUDE_ATTACHED_THREADS); } @@ -1400,17 +1403,19 @@ WITH RECURSIVE subtree(child_thread_id, parent_thread_id) AS ( builder.push(" AND threads.attached_thread_parent_id = "); builder.push_bind(parent.to_string()); } - None => {} + Some(crate::ThreadRelationFilter::IncludeAttached) | None => {} } let order_by_index = match (relation_filter, filters.cwd_filters) { // Relationship listings are expected to be much smaller than the global thread table. // Prefer the relationship index and sort the matches instead of scanning the // timestamp index until enough related threads happen to be found. - (Some(_), _) => OrderByIndex::Disabled, + (Some(relation), _) if relation != crate::ThreadRelationFilter::IncludeAttached => { + OrderByIndex::Disabled + } // Multi-cwd listing is supported but at the time of writing has no current use in production. // Preserve its query plan so the global timestamp index does not regress cwd filtering into a scan. - (None, Some(cwd_filters)) if cwd_filters.len() > 1 => OrderByIndex::Disabled, - (None, Some(_) | None) => OrderByIndex::Enabled, + (_, Some(cwd_filters)) if cwd_filters.len() > 1 => OrderByIndex::Disabled, + (_, Some(_) | None) => OrderByIndex::Enabled, }; push_thread_order_and_limit( builder, @@ -1523,7 +1528,7 @@ pub(super) fn push_thread_filters<'a>( builder, options, include_thread_id_tiebreaker, - /*include_empty_preview*/ false, + /*relation_filter*/ None, ); } @@ -1531,7 +1536,7 @@ fn push_thread_filters_with_preview<'a>( builder: &mut QueryBuilder, options: ThreadFilterOptions<'a>, include_thread_id_tiebreaker: bool, - include_empty_preview: bool, + relation_filter: Option, ) { let ThreadFilterOptions { archived_only, @@ -1551,8 +1556,18 @@ fn push_thread_filters_with_preview<'a>( } else { builder.push(" AND threads.archived = 0"); } - if !archived_only && !include_empty_preview && !matches!(section, Some(Some(_))) { - builder.push(" AND threads.preview <> ''"); + if !archived_only && !matches!(section, Some(Some(_))) { + match relation_filter { + None => { + builder.push(" AND threads.preview <> ''"); + } + Some(crate::ThreadRelationFilter::IncludeAttached) => { + builder.push( + " AND (threads.preview <> '' OR threads.attached_thread_parent_id IS NOT NULL)", + ); + } + Some(_) => {} + } } match section { Some(Some(section)) => { diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index a3e34aaeb0..7303edc797 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -1007,10 +1007,9 @@ impl ThreadStore for InMemoryThreadStore { page.items .retain(|thread| thread.parent_thread_id == Some(parent_thread_id)); } - Some(ThreadRelationFilter::AttachedThreadsOf(_)) => { - return Err(ThreadStoreError::Unsupported { - operation: "attached threads", - }); + Some(ThreadRelationFilter::AttachedThreadsOf(parent)) => { + page.items + .retain(|thread| thread.attached_to_thread_id == Some(parent)); } Some(ThreadRelationFilter::DescendantsOf(ancestor_thread_id)) => { let mut subtree = HashSet::from([ancestor_thread_id]); @@ -1033,7 +1032,10 @@ impl ThreadStore for InMemoryThreadStore { && subtree.contains(&thread.thread_id) }); } - None => {} + Some(ThreadRelationFilter::IncludeAttached) => {} + None => page + .items + .retain(|thread| thread.attached_to_thread_id.is_none()), } if let Some(section) = params.section.as_ref() { page.items.retain(|thread| { @@ -1132,6 +1134,7 @@ fn stored_thread_from_state( .and_then(|metadata| metadata.rollout_path.clone()) .or(rollout_path), forked_from_id: created.forked_from_id, + attached_to_thread_id: created.forked_from_id.filter(|_| created.attached_thread), parent_thread_id: created.parent_thread_id, preview: metadata .and_then(|metadata| metadata.preview.clone()) diff --git a/codex-rs/thread-store/src/local/helpers.rs b/codex-rs/thread-store/src/local/helpers.rs index a4f4d70552..1b5d0d74de 100644 --- a/codex-rs/thread-store/src/local/helpers.rs +++ b/codex-rs/thread-store/src/local/helpers.rs @@ -146,6 +146,7 @@ pub(super) fn stored_thread_from_rollout_item( extra_config: None, rollout_path: Some(rollout_path), forked_from_id: None, + attached_to_thread_id: item.attached_to_thread_id, parent_thread_id: item.parent_thread_id, preview, name: codex_state::is_guardian_review_source(&source) diff --git a/codex-rs/thread-store/src/local/list_threads.rs b/codex-rs/thread-store/src/local/list_threads.rs index 5abdecaf53..3d54f1b385 100644 --- a/codex-rs/thread-store/src/local/list_threads.rs +++ b/codex-rs/thread-store/src/local/list_threads.rs @@ -241,7 +241,13 @@ pub(super) async fn list_rollout_threads( sort_key: codex_rollout::ThreadSortKey, sort_direction: codex_rollout::SortDirection, ) -> ThreadStoreResult { - if params.relation_filter.is_some() || params.section.is_some() || params.project_id.is_some() { + if params.use_state_db_only + || params + .relation_filter + .is_some_and(ThreadRelationFilter::is_parent_scoped) + || params.section.is_some() + || params.project_id.is_some() + { let page = codex_rollout::state_db::list_threads_db( state_db.as_deref(), &config.sqlite, @@ -265,37 +271,9 @@ pub(super) async fn list_rollout_threads( return Ok(page.into()); } - let page = if params.use_state_db_only && params.archived { - RolloutRecorder::list_archived_threads_from_state_db( - state_db, - config, - params.page_size, - cursor, - sort_key, - sort_direction, - params.allowed_sources.as_slice(), - params.model_providers.as_deref(), - params.cwd_filters.as_deref(), - default_model_provider_id, - params.search_term.as_deref(), - ) - .await - } else if params.use_state_db_only { - RolloutRecorder::list_threads_from_state_db( - state_db, - config, - params.page_size, - cursor, - sort_key, - sort_direction, - params.allowed_sources.as_slice(), - params.model_providers.as_deref(), - params.cwd_filters.as_deref(), - default_model_provider_id, - params.search_term.as_deref(), - ) - .await - } else if params.archived { + let include_attached_threads = + params.relation_filter == Some(ThreadRelationFilter::IncludeAttached); + let page = if params.archived { RolloutRecorder::list_archived_threads( state_db, config, @@ -308,6 +286,7 @@ pub(super) async fn list_rollout_threads( params.cwd_filters.as_deref(), default_model_provider_id, params.search_term.as_deref(), + include_attached_threads, ) .await } else { @@ -323,6 +302,7 @@ pub(super) async fn list_rollout_threads( params.cwd_filters.as_deref(), default_model_provider_id, params.search_term.as_deref(), + include_attached_threads, ) .await }; @@ -333,6 +313,7 @@ pub(super) async fn list_rollout_threads( fn state_relation_filter(filter: ThreadRelationFilter) -> codex_state::ThreadRelationFilter { match filter { + ThreadRelationFilter::IncludeAttached => codex_state::ThreadRelationFilter::IncludeAttached, ThreadRelationFilter::DirectChildrenOf(id) => { codex_state::ThreadRelationFilter::DirectChildrenOf(id) } diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs index 9b1435c908..d943660c68 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -455,6 +455,7 @@ pub(super) fn stored_thread_from_state_metadata( thread_id: metadata.id, extra_config: None, rollout_path: Some(rollout_path), + attached_to_thread_id: metadata.attached_thread_parent_id, forked_from_id: None, parent_thread_id, preview, @@ -560,6 +561,10 @@ fn stored_thread_from_meta_line( thread_id: meta_line.meta.id, extra_config: None, rollout_path: Some(rollout_path), + attached_to_thread_id: meta_line + .meta + .forked_from_id + .filter(|_| meta_line.meta.attached_thread), forked_from_id: meta_line.meta.forked_from_id, parent_thread_id: meta_line.meta.parent_thread_id, preview: String::new(), @@ -864,6 +869,7 @@ mod tests { "Forked user message", Some("test-provider"), Some(parent_uuid), + /*attached_thread*/ false, ThreadHistoryMode::Legacy, ) .expect("forked session file"); diff --git a/codex-rs/thread-store/src/local/search_threads.rs b/codex-rs/thread-store/src/local/search_threads.rs index 6207811936..c9bd137038 100644 --- a/codex-rs/thread-store/src/local/search_threads.rs +++ b/codex-rs/thread-store/src/local/search_threads.rs @@ -106,7 +106,9 @@ pub(super) async fn search_threads( project_id: None, archived: params.archived, search_term: None, - relation_filter: None, + relation_filter: params + .include_attached_threads + .then_some(crate::ThreadRelationFilter::IncludeAttached), use_state_db_only: state_db.is_some(), }; let mut remaining_rollouts = matching_rollouts diff --git a/codex-rs/thread-store/src/local/search_threads_tests.rs b/codex-rs/thread-store/src/local/search_threads_tests.rs index 69737f4844..65148e2816 100644 --- a/codex-rs/thread-store/src/local/search_threads_tests.rs +++ b/codex-rs/thread-store/src/local/search_threads_tests.rs @@ -1,3 +1,7 @@ +use std::collections::HashMap; +use std::path::Path; +use std::path::PathBuf; + use chrono::Utc; use codex_protocol::ThreadId; use codex_protocol::protocol::SessionSource; @@ -10,10 +14,13 @@ use uuid::Uuid; use super::super::LocalThreadStore; use super::super::test_support::test_config; use super::super::test_support::write_session_file_with; +use super::super::test_support::write_session_file_with_fork; use super::ThreadSearchItem; use super::cursor_from_thread_search_item; +use crate::ListThreadsParams; use crate::SearchThreadsParams; use crate::SortDirection; +use crate::ThreadRelationFilter; use crate::ThreadSortKey; use crate::ThreadStore; @@ -141,6 +148,7 @@ async fn search_matches_selected_rollout_across_path_spellings_and_compression() ] { let page = store .search_threads(SearchThreadsParams { + include_attached_threads: false, account: None, page_size: 10, cursor: None, @@ -168,3 +176,147 @@ async fn search_matches_selected_rollout_across_path_spellings_and_compression() } } } + +async fn write_named_rollout(home: &Path, id: Uuid, attached_to: Option) -> PathBuf { + let path = write_session_file_with_fork( + home, + home.join("sessions/2025/01/03"), + "2025-01-03T12-00-00", + id, + "searchable message", + Some("test-provider"), + attached_to, + attached_to.is_some(), + ThreadHistoryMode::Legacy, + ) + .expect("write rollout"); + codex_rollout::append_thread_name(home, ThreadId::from_u128(id.as_u128()), "matching title") + .await + .expect("save thread name"); + path +} + +fn inclusive_title_search() -> ListThreadsParams { + ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + sort_direction: SortDirection::Desc, + allowed_sources: vec![SessionSource::Cli], + model_providers: None, + cwd_filters: None, + section: None, + project_id: None, + archived: false, + search_term: Some("matching title".to_string()), + relation_filter: Some(ThreadRelationFilter::IncludeAttached), + use_state_db_only: false, + } +} + +#[tokio::test] +async fn title_and_content_search_return_attached_parent_without_state_db() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let parent = Uuid::new_v4(); + let child = Uuid::new_v4(); + write_named_rollout(home.path(), child, Some(parent)).await; + let expected = vec![( + ThreadId::from_u128(child.as_u128()), + Some(ThreadId::from_u128(parent.as_u128())), + )]; + + let list = store + .list_threads(inclusive_title_search()) + .await + .expect("search titles on disk"); + assert_eq!( + list.items + .into_iter() + .map(|t| (t.thread_id, t.attached_to_thread_id)) + .collect::>(), + expected + ); + + let search = store + .search_threads(SearchThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + sort_direction: SortDirection::Desc, + allowed_sources: vec![SessionSource::Cli], + archived: false, + search_term: "searchable message".to_string(), + include_attached_threads: true, + account: None, + }) + .await + .expect("search messages on disk"); + assert_eq!( + search + .items + .into_iter() + .map(|hit| (hit.thread.thread_id, hit.thread.attached_to_thread_id)) + .collect::>(), + expected + ); +} + +#[tokio::test] +async fn inclusive_title_search_finds_unindexed_matches_when_filesystem_scanning_is_allowed() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let state_db = codex_state::StateRuntime::init( + config.sqlite.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("initialize state database"); + let parent = Uuid::new_v4(); + let child = Uuid::new_v4(); + let standalone = Uuid::new_v4(); + let parent_path = write_named_rollout(home.path(), parent, /*attached_to*/ None).await; + write_named_rollout(home.path(), standalone, /*attached_to*/ None).await; + write_named_rollout(home.path(), child, Some(parent)).await; + let parent_id = ThreadId::from_u128(parent.as_u128()); + let child_id = ThreadId::from_u128(child.as_u128()); + let mut row = codex_state::ThreadMetadataBuilder::new( + parent_id, + parent_path, + Utc::now(), + SessionSource::Cli, + ) + .build("test-provider"); + row.first_user_message = Some("searchable message".to_string()); + row.title = "matching title".to_string(); + state_db + .upsert_thread(&row) + .await + .expect("index only the parent"); + let store = LocalThreadStore::new(config, Some(state_db)); + let all = HashMap::from([ + (parent_id, None), + (child_id, Some(parent_id)), + (ThreadId::from_u128(standalone.as_u128()), None), + ]); + for (use_state_db_only, expected) in [ + (true, HashMap::from([(parent_id, None)])), + (false, all.clone()), + (true, all), + ] { + let page = store + .list_threads(ListThreadsParams { + use_state_db_only, + ..inclusive_title_search() + }) + .await + .expect("search titles"); + assert_eq!( + page.items + .into_iter() + .map(|t| (t.thread_id, t.attached_to_thread_id)) + .collect::>(), + expected + ); + } +} diff --git a/codex-rs/thread-store/src/local/test_support.rs b/codex-rs/thread-store/src/local/test_support.rs index e7b014b43b..ddaf7eece2 100644 --- a/codex-rs/thread-store/src/local/test_support.rs +++ b/codex-rs/thread-store/src/local/test_support.rs @@ -72,6 +72,7 @@ pub(super) fn write_session_file_with( first_user_message, model_provider, /*forked_from_id*/ None, + /*attached_thread*/ false, history_mode, ) } @@ -85,6 +86,7 @@ pub(super) fn write_session_file_with_fork( first_user_message: &str, model_provider: Option<&str>, forked_from_id: Option, + attached_thread: bool, history_mode: ThreadHistoryMode, ) -> std::io::Result { fs::create_dir_all(&day_dir)?; @@ -97,6 +99,7 @@ pub(super) fn write_session_file_with_fork( "session_id": uuid, "id": uuid, "forked_from_id": forked_from_id, + "attached_thread": attached_thread, "timestamp": ts, "cwd": root, "originator": "test_originator", diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 66c26692df..02979d0f68 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -337,6 +337,8 @@ pub enum SortDirection { /// Parent relationship used to filter thread listings. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum ThreadRelationFilter { + /// Include attached threads alongside ordinary threads for search. + IncludeAttached, /// Return only threads whose immediate parent is the given thread. DirectChildrenOf(ThreadId), /// Return every thread transitively descended from the given thread. @@ -345,6 +347,16 @@ pub enum ThreadRelationFilter { AttachedThreadsOf(ThreadId), } +impl ThreadRelationFilter { + /// Whether the caller requested relatives of one particular thread. + pub fn is_parent_scoped(self) -> bool { + match self { + Self::IncludeAttached => false, + Self::DirectChildrenOf(_) | Self::DescendantsOf(_) | Self::AttachedThreadsOf(_) => true, + } + } +} + /// Parameters for listing threads. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ListThreadsParams { @@ -383,6 +395,8 @@ pub struct ListThreadsParams { /// Parameters for searching thread content. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct SearchThreadsParams { + #[serde(default)] + pub include_attached_threads: bool, /// The current login when searching account-scoped local history. #[serde(default, skip_serializing_if = "Option::is_none")] pub account: Option, @@ -660,6 +674,9 @@ pub struct StoredThread { pub rollout_path: Option, /// Source thread id when this thread was forked from another thread. pub forked_from_id: Option, + /// Parent that owns this persisted attached thread, independent of fork or spawn lineage. + #[serde(default)] + pub attached_to_thread_id: Option, /// The ID of the parent thread. This will only be set if this thread is a subagent. pub parent_thread_id: Option, /// Best available user-facing preview, usually the first user message. diff --git a/codex-rs/tui/src/analytics/chats.rs b/codex-rs/tui/src/analytics/chats.rs index f9dd8c68f2..9c32e57835 100644 --- a/codex-rs/tui/src/analytics/chats.rs +++ b/codex-rs/tui/src/analytics/chats.rs @@ -144,6 +144,7 @@ pub(super) async fn roots( pub(super) fn list_params() -> ThreadListParams { ThreadListParams { + include_attached_threads: false, current_account_only: true, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/tui/src/app/agent_picker.rs b/codex-rs/tui/src/app/agent_picker.rs index 1132d640f1..681d123a68 100644 --- a/codex-rs/tui/src/app/agent_picker.rs +++ b/codex-rs/tui/src/app/agent_picker.rs @@ -40,6 +40,7 @@ impl App { .request_typed::(ClientRequest::ThreadList { request_id: RequestId::String(Uuid::new_v4().to_string()), params: ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/tui/src/app/agents_overview_discovery.rs b/codex-rs/tui/src/app/agents_overview_discovery.rs index 8c951fbe2c..41464b3ebb 100644 --- a/codex-rs/tui/src/app/agents_overview_discovery.rs +++ b/codex-rs/tui/src/app/agents_overview_discovery.rs @@ -48,6 +48,7 @@ pub(crate) async fn list_pinned_threads( .request_typed::(ClientRequest::ThreadList { request_id: RequestId::String(Uuid::new_v4().to_string()), params: ThreadListParams { + include_attached_threads: false, current_account_only: true, excluded_thread_ids: None, originators: None, @@ -193,6 +194,7 @@ impl SourcePage { .request_typed::(ClientRequest::ThreadList { request_id: RequestId::String(Uuid::new_v4().to_string()), params: ThreadListParams { + include_attached_threads: false, current_account_only: true, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/tui/src/app/agents_overview_tests.rs b/codex-rs/tui/src/app/agents_overview_tests.rs index 114e685a8e..287430e457 100644 --- a/codex-rs/tui/src/app/agents_overview_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_tests.rs @@ -269,6 +269,7 @@ pub(in crate::app) fn overview_thread( daybreak_enabled: None, session_id: parent_thread_id.unwrap_or(thread_id).to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: name.to_string(), ephemeral: false, diff --git a/codex-rs/tui/src/app/loaded_threads.rs b/codex-rs/tui/src/app/loaded_threads.rs index 57c9c6b336..75d3ed5ba5 100644 --- a/codex-rs/tui/src/app/loaded_threads.rs +++ b/codex-rs/tui/src/app/loaded_threads.rs @@ -141,6 +141,7 @@ mod tests { extra: None, session_id: thread_id.to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: String::new(), ephemeral: false, diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index f39a812d76..cda3ed85b0 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -7085,6 +7085,7 @@ async fn inactive_thread_started_notification_initializes_replay_session() -> Re extra: None, session_id: agent_thread_id.to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: "agent thread".to_string(), ephemeral: false, @@ -7191,6 +7192,7 @@ async fn inactive_thread_started_notification_preserves_primary_model_when_path_ extra: None, session_id: agent_thread_id.to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: "agent thread".to_string(), ephemeral: false, @@ -7262,6 +7264,7 @@ async fn thread_read_session_state_does_not_reuse_primary_permission_profile() { extra: None, session_id: read_thread_id.to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: "read thread".to_string(), ephemeral: false, diff --git a/codex-rs/tui/src/app/tests/startup.rs b/codex-rs/tui/src/app/tests/startup.rs index 7a91a3a2fb..4954bf3ff7 100644 --- a/codex-rs/tui/src/app/tests/startup.rs +++ b/codex-rs/tui/src/app/tests/startup.rs @@ -1304,6 +1304,7 @@ async fn known_thread_started_preserves_session_without_reading_unmaterialized_r extra: None, session_id: thread_id.to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: String::new(), ephemeral: false, diff --git a/codex-rs/tui/src/app/thread_session_state.rs b/codex-rs/tui/src/app/thread_session_state.rs index b5596528f2..983b95b25a 100644 --- a/codex-rs/tui/src/app/thread_session_state.rs +++ b/codex-rs/tui/src/app/thread_session_state.rs @@ -418,6 +418,7 @@ mod tests { extra: None, session_id: read_thread_id.to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: "read thread".to_string(), ephemeral: false, diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 73351e555f..d83055f5d4 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -4912,6 +4912,7 @@ mod tests { extra: None, session_id: ThreadId::new().to_string(), forked_from_id: Some(forked_from_id.to_string()), + attached_to_thread_id: None, parent_thread_id: None, preview: "hello".to_string(), ephemeral: false, diff --git a/codex-rs/tui/src/app_server_session/side_chat_history.rs b/codex-rs/tui/src/app_server_session/side_chat_history.rs index fd018d5948..72a92e2a2d 100644 --- a/codex-rs/tui/src/app_server_session/side_chat_history.rs +++ b/codex-rs/tui/src/app_server_session/side_chat_history.rs @@ -125,6 +125,7 @@ impl AppServerSession { } let query = ThreadListParams { attached_to_thread_id: Some(parent.to_string()), + include_attached_threads: false, current_account_only: true, archived: Some(false), model_providers: Some(Vec::new()), diff --git a/codex-rs/tui/src/dynamic_tools.rs b/codex-rs/tui/src/dynamic_tools.rs index 7e753f16fc..e06585de15 100644 --- a/codex-rs/tui/src/dynamic_tools.rs +++ b/codex-rs/tui/src/dynamic_tools.rs @@ -351,6 +351,7 @@ async fn execute_inner( request(&handle, |request_id| ClientRequest::ThreadList { request_id, params: ThreadListParams { + include_attached_threads: false, current_account_only: true, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index 2d124a4f3b..52d6aefaeb 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -939,6 +939,7 @@ fn latest_session_lookup_params( lookup_mode: LatestSessionLookupMode, ) -> ThreadListParams { ThreadListParams { + include_attached_threads: false, current_account_only: true, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/tui/src/named_session_lookup.rs b/codex-rs/tui/src/named_session_lookup.rs index 06529b6b23..40f724e837 100644 --- a/codex-rs/tui/src/named_session_lookup.rs +++ b/codex-rs/tui/src/named_session_lookup.rs @@ -64,6 +64,7 @@ pub(super) async fn lookup( loop { let response = app_server .thread_list(ThreadListParams { + include_attached_threads: false, current_account_only: false, excluded_thread_ids: None, attached_to_thread_id: None, diff --git a/codex-rs/tui/src/resume_picker.rs b/codex-rs/tui/src/resume_picker.rs index d2d45d3023..a9d7fbce57 100644 --- a/codex-rs/tui/src/resume_picker.rs +++ b/codex-rs/tui/src/resume_picker.rs @@ -2515,6 +2515,7 @@ fn thread_list_params( use_state_db_only: bool, ) -> ThreadListParams { ThreadListParams { + include_attached_threads: false, current_account_only: true, excluded_thread_ids: None, attached_to_thread_id: None, @@ -7451,6 +7452,7 @@ session_picker_view = "dense" extra: None, session_id: thread_id.to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: String::from("remote thread"), ephemeral: false, @@ -7498,6 +7500,7 @@ session_picker_view = "dense" extra: None, session_id: thread_id.to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: String::from("preview"), ephemeral: false, @@ -7600,6 +7603,7 @@ session_picker_view = "dense" extra: None, session_id: thread_id.to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: String::from("preview"), ephemeral: false, @@ -7679,6 +7683,7 @@ session_picker_view = "dense" extra: None, session_id: thread_id.to_string(), forked_from_id: None, + attached_to_thread_id: None, parent_thread_id: None, preview: String::from("preview"), ephemeral: false, diff --git a/codex-rs/tui/src/task_mentions.rs b/codex-rs/tui/src/task_mentions.rs index c7d8e8e148..2d3cc54906 100644 --- a/codex-rs/tui/src/task_mentions.rs +++ b/codex-rs/tui/src/task_mentions.rs @@ -73,6 +73,7 @@ pub(crate) fn spawn_search( handle.request_typed::(ClientRequest::ThreadSearch { request_id: RequestId::String(format!("task-search-{request_generation}")), params: ThreadSearchParams { + include_attached_threads: false, current_account_only: true, cursor: None, limit: Some(MAX_SEARCH_RESULTS), @@ -91,6 +92,7 @@ pub(crate) fn spawn_search( "task-title-search-{request_generation}" )), params: ThreadListParams { + include_attached_threads: false, current_account_only: true, excluded_thread_ids: None, attached_to_thread_id: None,