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 c4b50dc236..c52bd5525b 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/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 0592a2f3ad..4d62ef4b2d 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ 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 a7a76234f3..79422eba09 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -623,6 +623,12 @@ pub struct ThreadForkParams { pub developer_instructions: Option, #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub ephemeral: bool, + /// Persist the fork as an attached thread of its resolved source, hidden from ordinary + /// lists. The source cannot be deleted while attached threads exist. Requires local storage; + /// incompatible with ephemeral and deferGoalContinuation. Omitted or false creates an ordinary fork. + #[experimental("thread/fork.attachToSource")] + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub attach_to_source: bool, /// Optional client-supplied analytics source classification for this forked thread. #[ts(optional = nullable)] pub thread_source: Option, @@ -1483,6 +1489,12 @@ pub struct ThreadListParams { #[experimental("thread/list.ancestorThreadId")] #[ts(optional = nullable)] pub ancestor_thread_id: Option, + /// Return only persistent threads attached to this thread, including empty threads. + /// Mutually exclusive with parentThreadId and ancestorThreadId. + /// Omitting this filter keeps attached threads hidden from ordinary listings. + #[experimental("thread/list.attachedToThreadId")] + #[ts(optional = nullable)] + pub attached_to_thread_id: Option, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] diff --git a/codex-rs/app-server-test-client/src/lib.rs b/codex-rs/app-server-test-client/src/lib.rs index 26aeed921f..5db6d881ae 100644 --- a/codex-rs/app-server-test-client/src/lib.rs +++ b/codex-rs/app-server-test-client/src/lib.rs @@ -1168,6 +1168,7 @@ async fn thread_list(endpoint: &Endpoint, config_overrides: &[String], limit: u3 let response = client.thread_list(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(limit), diff --git a/codex-rs/app-server/src/external_agent_migration/session_importer.rs b/codex-rs/app-server/src/external_agent_migration/session_importer.rs index 13fbb93627..4a05a1d078 100644 --- a/codex-rs/app-server/src/external_agent_migration/session_importer.rs +++ b/codex-rs/app-server/src/external_agent_migration/session_importer.rs @@ -455,6 +455,7 @@ impl ExternalAgentSessionImporter { }; let now = Utc::now(); let create_params = CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/app-server/src/request_processors/thread_delete.rs b/codex-rs/app-server/src/request_processors/thread_delete.rs index 7b014daa21..7fd33e0997 100644 --- a/codex-rs/app-server/src/request_processors/thread_delete.rs +++ b/codex-rs/app-server/src/request_processors/thread_delete.rs @@ -40,6 +40,16 @@ impl ThreadRequestProcessor { self.validate_root_thread_delete(thread_id, thread_ids.len() > 1) .await?; + if let Some(store) = self + .thread_store + .as_any() + .downcast_ref::() + { + store + .ensure_no_attached_threads(&thread_ids) + .await + .map_err(thread_store_delete_error)?; + } for thread_id_to_delete in thread_ids.iter().copied() { self.prepare_thread_for_delete(thread_id_to_delete).await?; } 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 925feda55c..ffade96e3c 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2697,6 +2697,7 @@ impl ThreadRequestProcessor { search_term, parent_thread_id, ancestor_thread_id, + attached_to_thread_id, } = params; let excluded_thread_ids = excluded_thread_ids.unwrap_or_default(); if excluded_thread_ids.len() > THREAD_LIST_MAX_EXCLUDED_IDS { @@ -2740,21 +2741,42 @@ impl ThreadRequestProcessor { } } let cwd_filters = normalize_thread_list_cwd_filters(cwd)?; - let relation_filter = match (parent_thread_id, ancestor_thread_id) { - (Some(_), Some(_)) => { + let relation_filter = match (parent_thread_id, ancestor_thread_id, attached_to_thread_id) { + (Some(_), Some(_), None) => { return Err(invalid_request( "parentThreadId and ancestorThreadId are mutually exclusive", )); } - (Some(parent_thread_id), None) => Some(StoreThreadRelationFilter::DirectChildrenOf( - ThreadId::from_string(&parent_thread_id) - .map_err(|err| invalid_request(format!("invalid parent thread id: {err}")))?, - )), - (None, Some(ancestor_thread_id)) => Some(StoreThreadRelationFilter::DescendantsOf( - ThreadId::from_string(&ancestor_thread_id) - .map_err(|err| invalid_request(format!("invalid ancestor thread id: {err}")))?, - )), - (None, None) => None, + (Some(parent_thread_id), None, None) => { + Some(StoreThreadRelationFilter::DirectChildrenOf( + ThreadId::from_string(&parent_thread_id).map_err(|err| { + invalid_request(format!("invalid parent thread id: {err}")) + })?, + )) + } + (None, Some(ancestor_thread_id), None) => { + Some(StoreThreadRelationFilter::DescendantsOf( + ThreadId::from_string(&ancestor_thread_id).map_err(|err| { + invalid_request(format!("invalid ancestor thread id: {err}")) + })?, + )) + } + (None, None, Some(parent)) => { + if self.state_db.is_none() || !self.thread_store.as_any().is::() { + return Err(unsupported_thread_store_operation("attached threads")); + } + Some(StoreThreadRelationFilter::AttachedThreadsOf( + ThreadId::from_string(&parent).map_err(|err| { + invalid_request(format!("invalid attached thread parent id: {err}")) + })?, + )) + } + (_, _, Some(_)) => { + return Err(invalid_request( + "attachedToThreadId cannot be combined with parentThreadId or ancestorThreadId", + )); + } + (None, None, None) => None, }; let requested_page_size = limit @@ -5269,11 +5291,22 @@ impl ThreadRequestProcessor { base_instructions, developer_instructions, ephemeral, + attach_to_source, thread_source, exclude_turns, defer_goal_continuation, } = params; let include_turns = !exclude_turns; + if attach_to_source { + if ephemeral || defer_goal_continuation { + return Err(invalid_request( + "attachToSource cannot be combined with ephemeral or deferGoalContinuation", + )); + } + if self.state_db.is_none() || !self.thread_store.as_any().is::() { + return Err(unsupported_thread_store_operation("attached threads")); + } + } if sandbox.is_some() && permissions.is_some() { return Err(invalid_request( "`permissions` cannot be combined with `sandbox`", @@ -5645,6 +5678,7 @@ impl ThreadRequestProcessor { }; let fork_options = StartThreadOptions { + attached_thread: attach_to_source, thread_source, parent_trace, reserved_thread_id, @@ -5877,9 +5911,11 @@ impl ThreadRequestProcessor { .await; } - self.outgoing - .send_server_notification(ServerNotification::ThreadStarted(notif)) - .await; + if !attach_to_source { + self.outgoing + .send_server_notification(ServerNotification::ThreadStarted(notif)) + .await; + } if inherited_goal { self.thread_goal_processor .emit_thread_goal_snapshot(thread_id) diff --git a/codex-rs/app-server/tests/common/rollout.rs b/codex-rs/app-server/tests/common/rollout.rs index 75a48b5987..a7c3222929 100644 --- a/codex-rs/app-server/tests/common/rollout.rs +++ b/codex-rs/app-server/tests/common/rollout.rs @@ -253,6 +253,7 @@ fn create_fake_rollout_with_source_and_parent_thread_id( // Build JSONL lines let meta = SessionMeta { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id, @@ -349,6 +350,7 @@ pub fn create_fake_rollout_with_text_elements( // Build JSONL lines let meta = SessionMeta { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: conversation_id.into(), diff --git a/codex-rs/app-server/tests/suite/conversation_summary.rs b/codex-rs/app-server/tests/suite/conversation_summary.rs index 5c55734842..aaeed4ec19 100644 --- a/codex-rs/app-server/tests/suite/conversation_summary.rs +++ b/codex-rs/app-server/tests/suite/conversation_summary.rs @@ -122,6 +122,7 @@ async fn get_conversation_summary_by_thread_id_reads_pathless_store_thread() -> let thread_id = ThreadId::from_string("00000000-0000-4000-8000-000000000125")?; store .create_thread(CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), 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 new file mode 100644 index 0000000000..5f1b053276 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/attached_thread_tests.rs @@ -0,0 +1,339 @@ +//! Attached-thread fork and discovery after restart and SQLite recovery, including pre-database parents. +use anyhow::Result; +use app_test_support::MockResponsesConfig; +use app_test_support::TestAppServer; +use app_test_support::create_fake_paginated_rollout; +use app_test_support::create_fake_rollout; +use app_test_support::create_mock_responses_server_repeating_assistant; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadDeleteParams; +use codex_app_server_protocol::ThreadDeleteResponse; +use codex_app_server_protocol::ThreadForkParams; +use codex_app_server_protocol::ThreadForkResponse; +use codex_app_server_protocol::ThreadHistoryMode; +use codex_app_server_protocol::ThreadListResponse; +use codex_app_server_protocol::ThreadLoadedListParams; +use codex_app_server_protocol::ThreadLoadedListResponse; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::UserInput; +use codex_state::SqliteConfig; +use codex_utils_absolute_path::test_support::PathExt; +use pretty_assertions::assert_eq; +use serde_json::json; +use tempfile::TempDir; +use test_case::test_case; + +#[derive(Clone, Copy)] +enum Recovery { + Restart, + Missing, + Corrupt, +} + +#[test_case(ThreadHistoryMode::Legacy, Recovery::Restart; "legacy restart")] +#[test_case(ThreadHistoryMode::Paginated, Recovery::Restart; "paginated restart")] +#[test_case(ThreadHistoryMode::Legacy, Recovery::Missing; "legacy missing database")] +#[test_case(ThreadHistoryMode::Paginated, Recovery::Corrupt; "paginated corrupt database")] +#[tokio::test] +async fn saved_attached_thread_upgrade_restart_and_delete( + mode: ThreadHistoryMode, + recovery: Recovery, +) -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()).write(home.path())?; + // Both parents were persisted before this app-server (and its SQLite migration) starts. + let create_rollout = match mode { + ThreadHistoryMode::Legacy => create_fake_rollout, + ThreadHistoryMode::Paginated => create_fake_paginated_rollout, + }; + let parent = create_rollout( + home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + "BLUE HERON 72", + Some("mock_provider"), + /*git_info*/ None, + )?; + let other = create_rollout( + home.path(), + "2025-01-06T12-00-00", + "2025-01-06T12:00:00Z", + "unrelated", + Some("mock_provider"), + /*git_info*/ None, + )?; + let mut mcp = TestAppServer::builder() + .with_codex_home(home.path()) + .without_auto_env() + .build_initialized() + .await?; + let mut children = Vec::new(); + for _ in 0..2 { + let id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: parent.clone(), + attach_to_source: true, + exclude_turns: true, + ..Default::default() + }) + .await?; + let fork: ThreadForkResponse = mcp.read_response(id).await?; + assert!(!fork.thread.ephemeral); + assert_eq!(fork.thread.forked_from_id, Some(parent.clone())); + children.push(fork.thread.id); + } + // Newer attached threads fill the raw scan page. Listing must refill across hidden + // results, with and without title filters and in both scan directions. + for (params, expected) in [ + ( + json!({"limit": 1, "searchTerm": "BLUE HERON"}), + vec![&parent], + ), + ( + json!({"limit": 1, "searchTerm": "BLUE HERON", "useStateDbOnly": true}), + vec![&parent], + ), + ( + json!({"limit": 2, "modelProviders": ["mock_provider"]}), + vec![&other, &parent], + ), + ( + json!({"limit": 2, "modelProviders": ["mock_provider"], "sortDirection": "asc"}), + vec![&parent, &other], + ), + ( + json!({"limit": 2, "useStateDbOnly": true}), + vec![&other, &parent], + ), + // The existing parent filter continues to mean spawned children. + (json!({"parentThreadId": parent}), vec![]), + ] { + let id = mcp.send_raw_request("thread/list", Some(params)).await?; + let ordinary: ThreadListResponse = mcp.read_response(id).await?; + assert_eq!( + ordinary.data.iter().map(|t| &t.id).collect::>(), + expected + ); + } + let sqlite = SqliteConfig::new_for_testing(home.path().canonicalize()?.as_path().abs()); + assert!(mcp.shutdown_gracefully().await?.success()); + // No child has taken a turn yet; its attachment must still survive a restart. + match recovery { + Recovery::Restart => {} + Recovery::Missing => tokio::fs::remove_file(sqlite.state_db_path()).await?, + Recovery::Corrupt => { + tokio::fs::write(sqlite.state_db_path(), b"damaged state database").await? + } + } + let mut mcp = TestAppServer::builder() + .with_codex_home(home.path()) + .without_auto_env() + .build_initialized() + .await?; + let id = mcp + .send_raw_request( + "thread/list", + Some(json!({"useStateDbOnly": true, "limit": 2})), + ) + .await?; + let ordinary: ThreadListResponse = mcp.read_response(id).await?; + assert_eq!( + ordinary.data.iter().map(|t| &t.id).collect::>(), + vec![&other, &parent] + ); + let mut cursor = None; + let mut found = Vec::new(); + loop { + let id = mcp + .send_raw_request( + "thread/list", + Some(json!({ + "attachedToThreadId": parent, "limit": 1, "cursor": cursor + })), + ) + .await?; + let page: ThreadListResponse = mcp.read_response(id).await?; + assert_eq!(page.data.len(), 1); + found.push(page.data[0].id.clone()); + cursor = page.next_cursor; + if cursor.is_none() { + break; + } + assert!( + found.len() <= children.len(), + "attached thread cursor must advance" + ); + } + found.sort(); + children.sort(); + assert_eq!(found, children); + let id = mcp + .send_raw_request("thread/list", Some(json!({"attachedToThreadId": other}))) + .await?; + let other_list: ThreadListResponse = mcp.read_response(id).await?; + assert!(other_list.data.is_empty()); + let id = mcp + .send_raw_request( + "thread/resume", + Some(json!({ + "threadId": children[0], "excludeTurns": true, + "developerInstructions": "Follow BIRCH-849." + })), + ) + .await?; + let resumed: ThreadResumeResponse = mcp.read_response(id).await?; + assert_eq!(resumed.thread.id, children[0]); + // Ordinary forks do not inherit attachment. + let id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: children[0].clone(), + ..Default::default() + }) + .await?; + let independent: ThreadForkResponse = mcp.read_response(id).await?; + let id = mcp + .send_raw_request( + "thread/list", + Some(json!({"attachedToThreadId": children[0]})), + ) + .await?; + let attached: ThreadListResponse = mcp.read_response(id).await?; + assert!(attached.data.is_empty()); + let id = mcp + .send_thread_delete_request(ThreadDeleteParams { + thread_id: independent.thread.id, + }) + .await?; + let _: ThreadDeleteResponse = mcp.read_response(id).await?; + let id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: children[0].clone(), + input: vec![UserInput::Text { + text: "continue".into(), + text_elements: vec![], + }], + ..Default::default() + }) + .await?; + let _: TurnStartResponse = mcp.read_response(id).await?; + mcp.read_stream_until_notification_message("turn/completed") + .await?; + let requests = server.received_requests().await.expect("response requests"); + let input = requests + .iter() + .find(|r| r.url.path().ends_with("/responses")) + .expect("resumed attached thread sends a model request") + .body_json::()?["input"] + .to_string(); + assert!( + input.contains("BIRCH-849"), + "resume applies client instructions" + ); + let id = mcp + .send_raw_request("thread/resume", Some(json!({"threadId": parent}))) + .await?; + let _: ThreadResumeResponse = mcp.read_response(id).await?; + let id = mcp + .send_thread_delete_request(ThreadDeleteParams { + thread_id: parent.clone(), + }) + .await?; + let error = mcp + .read_stream_until_error_message(RequestId::Integer(id)) + .await?; + assert!( + error + .error + .message + .contains("delete the attached thread first") + ); + let id = mcp + .send_thread_loaded_list_request(ThreadLoadedListParams::default()) + .await?; + let loaded: ThreadLoadedListResponse = mcp.read_response(id).await?; + assert!( + loaded.data.contains(&parent), + "rejection must leave the parent loaded" + ); + for child in children { + let id = mcp + .send_thread_delete_request(ThreadDeleteParams { thread_id: child }) + .await?; + let _: ThreadDeleteResponse = mcp.read_response(id).await?; + } + let id = mcp + .send_thread_delete_request(ThreadDeleteParams { thread_id: parent }) + .await?; + let _: ThreadDeleteResponse = mcp.read_response(id).await?; + let id = mcp.send_raw_request("thread/list", Some(json!({}))).await?; + let ordinary: ThreadListResponse = mcp.read_response(id).await?; + assert_eq!( + ordinary.data.iter().map(|t| &t.id).collect::>(), + vec![&other] + ); + Ok(()) +} + +#[tokio::test] +async fn saved_attached_thread_rejects_conflicting_fork_and_discovery_filters() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()).write(home.path())?; + let parent = create_fake_rollout( + home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + "parent", + Some("mock_provider"), + /*git_info*/ None, + )?; + let mut mcp = TestAppServer::builder() + .with_codex_home(home.path()) + .without_auto_env() + .build_initialized() + .await?; + for (method, params, message) in [ + ( + "thread/fork", + json!({"threadId": parent, "attachToSource": true, "ephemeral": true}), + "attachToSource cannot be combined", + ), + ( + "thread/list", + json!({"attachedToThreadId": parent, "parentThreadId": parent}), + "attachedToThreadId cannot be combined", + ), + ( + "thread/list", + json!({"attachedToThreadId": parent, "ancestorThreadId": parent}), + "attachedToThreadId cannot be combined", + ), + ( + "thread/fork", + json!({"threadId": parent, "attachToSource": true, "deferGoalContinuation": true}), + "attachToSource cannot be combined", + ), + ] { + let id = mcp.send_raw_request(method, Some(params)).await?; + let error = mcp + .read_stream_until_error_message(RequestId::Integer(id)) + .await?; + assert!( + error.error.message.starts_with(message), + "{:?}", + error.error + ); + } + let id = mcp + .send_raw_request("thread/list", Some(json!({"attachedToThreadId": parent}))) + .await?; + let page: ThreadListResponse = mcp.read_response(id).await?; + assert!( + page.data.is_empty(), + "failed forks must leave no discoverable attached thread" + ); + Ok(()) +} 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 fe05440a40..08621c0d7b 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 { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: None, @@ -1976,6 +1977,7 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { let request_id = mcp .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: None, @@ -2168,6 +2170,7 @@ required = true let request_id = mcp .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: None, @@ -2255,6 +2258,7 @@ async fn external_agent_config_import_accepts_detected_session_payload_after_res let request_id = mcp .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: None, @@ -2339,6 +2343,7 @@ async fn external_agent_config_import_skips_already_imported_session_versions() let request_id = mcp .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: None, @@ -2469,6 +2474,7 @@ async fn external_agent_config_import_returns_before_background_session_import_f let request_id = mcp .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: None, @@ -2587,6 +2593,7 @@ async fn external_agent_config_import_compacts_huge_session_before_first_follow_ let request_id = mcp .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: None, diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 21df247cf3..6d1e77143f 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -10,6 +10,8 @@ mod app_installed; mod app_list; mod app_read; mod application_network; +#[path = "attached_thread_tests.rs"] +mod attached_thread; mod attestation; #[path = "auth_storage_originator_tests.rs"] mod auth_storage_originator; diff --git a/codex-rs/app-server/tests/suite/v2/projects.rs b/codex-rs/app-server/tests/suite/v2/projects.rs index c8d2a4a989..56c4d6e9d8 100644 --- a/codex-rs/app-server/tests/suite/v2/projects.rs +++ b/codex-rs/app-server/tests/suite/v2/projects.rs @@ -101,6 +101,7 @@ async fn projects_list_by_recency_and_preserve_metadata_timestamps() -> Result<( request_id, params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -326,6 +327,7 @@ async fn projects_persist_and_assign_threads() -> Result<()> { request_id, params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -374,6 +376,7 @@ async fn projects_persist_and_assign_threads() -> Result<()> { request_id, params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -425,6 +428,7 @@ async fn projects_persist_and_assign_threads() -> Result<()> { request_id, params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -621,6 +625,7 @@ async fn projects_persist_and_assign_threads() -> Result<()> { request_id, params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -713,6 +718,7 @@ async fn deleted_project_is_dropped_before_first_durable_thread_persistence() -> request_id, params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -904,6 +910,7 @@ async fn projects_validate_filters_cursors_and_sqlite_less_assignment() -> Resul let request_id = server .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -991,6 +998,7 @@ async fn projects_validate_filters_cursors_and_sqlite_less_assignment() -> Resul let list_id = unsupported_projects .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), 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 c7b7545d6e..90a6e7fc61 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 @@ -27,6 +27,7 @@ use codex_app_server::in_process::InProcessServerEvent; use codex_app_server::in_process::InProcessStartArgs; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::InitializeCapabilities; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::RequestId; @@ -190,6 +191,53 @@ async fn thread_start_defaults_to_legacy_without_history_list_support() -> Resul Ok(()) } +#[tokio::test] +async fn attached_threads_require_store_owned_persistence() -> Result<()> { + let codex_home = TempDir::new()?; + let store_id = Uuid::new_v4().to_string(); + create_config_toml_with_thread_store(codex_home.path(), "http://127.0.0.1:1", &store_id)?; + let _in_memory_store = InMemoryThreadStoreId { store_id }; + let client = start_in_process_server_with_capabilities( + codex_home.path(), + Some(InitializeCapabilities { + experimental_api: true, + ..Default::default() + }), + ) + .await?; + // The store, not the experiment gate, must reject it without writing to local storage. + for request in [ + ClientRequest::ThreadFork { + request_id: RequestId::Integer(1), + params: codex_app_server_protocol::ThreadForkParams { + thread_id: Uuid::new_v4().to_string(), + attach_to_source: true, + ..Default::default() + }, + }, + ClientRequest::ThreadList { + request_id: RequestId::Integer(2), + params: serde_json::from_value(serde_json::json!({ + "attachedToThreadId": Uuid::new_v4().to_string() + }))?, + }, + ] { + let error = client + .request(request) + .await? + .expect_err("attached threads require local storage"); + assert_eq!(error.code, -32601); + assert!( + error.message.contains("attached threads"), + "{}", + error.message + ); + } + client.shutdown().await?; + assert_no_local_persistence_artifacts(codex_home.path())?; + Ok(()) +} + #[tokio::test] async fn thread_list_repeated_cursor_returns_internal_error() -> Result<()> { let codex_home = TempDir::new()?; @@ -207,6 +255,7 @@ async fn thread_list_repeated_cursor_returns_internal_error() -> Result<()> { request_id: RequestId::Integer(1), params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(1), @@ -232,6 +281,7 @@ async fn thread_list_repeated_cursor_returns_internal_error() -> Result<()> { let thread_id = ThreadId::new(); thread_store .create_thread(StoreCreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), @@ -264,6 +314,7 @@ async fn thread_list_repeated_cursor_returns_internal_error() -> Result<()> { request_id: RequestId::Integer(2), params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: Some("repeated-cursor".to_string()), limit: Some(1), @@ -446,6 +497,7 @@ async fn thread_delete_with_non_local_thread_store_does_not_create_local_persist request_id: RequestId::Integer(3), params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -475,6 +527,7 @@ async fn thread_delete_with_non_local_thread_store_does_not_create_local_persist let unloaded_thread_id = ThreadId::from_string(&Uuid::new_v4().to_string())?; thread_store .create_thread(StoreCreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: unloaded_thread_id.into(), @@ -547,7 +600,12 @@ async fn cold_thread_resume_rechecks_non_local_history_after_config_load() -> Re let thread_store = InMemoryThreadStore::for_id(store_id.clone()); let _in_memory_store = InMemoryThreadStoreId { store_id }; - let mut client = start_in_process_client(config.clone(), loader_overrides.clone()).await?; + let mut client = start_in_process_client( + config.clone(), + loader_overrides.clone(), + /*capabilities*/ None, + ) + .await?; let response = client .request(ClientRequest::ThreadStart { request_id: RequestId::Integer(1), @@ -588,7 +646,7 @@ async fn cold_thread_resume_rechecks_non_local_history_after_config_load() -> Re .await??; client.shutdown().await?; - let client = start_in_process_client(config, loader_overrides).await?; + let client = start_in_process_client(config, loader_overrides, /*capabilities*/ None).await?; let reads_before_resume = thread_store.calls().await.read_thread_with_history; // The in-memory store is pathless, so resume currently fails later while // assembling the response. Reuse the probe within each attempt, but read it @@ -610,6 +668,13 @@ async fn cold_thread_resume_rechecks_non_local_history_after_config_load() -> Re } async fn start_in_process_server(codex_home: &Path) -> Result { + start_in_process_server_with_capabilities(codex_home, /*capabilities*/ None).await +} + +async fn start_in_process_server_with_capabilities( + codex_home: &Path, + capabilities: Option, +) -> Result { let loader_overrides = LoaderOverrides::without_managed_config_for_tests(); let config = Arc::new( ConfigBuilder::default() @@ -620,12 +685,13 @@ async fn start_in_process_server(codex_home: &Path) -> Result, loader_overrides: LoaderOverrides, + capabilities: Option, ) -> std::io::Result { in_process::start(InProcessStartArgs { arg0_paths: Arg0DispatchPaths::default(), @@ -649,7 +715,7 @@ async fn start_in_process_client( title: None, version: "0.1.0".to_string(), }, - capabilities: None, + capabilities, }, channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY, }) 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 a3e94c74c2..1a812731d5 100644 --- a/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs +++ b/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs @@ -160,6 +160,7 @@ async fn state_recovery_restores_saved_threads() -> Result<()> { let request = server .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: 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 980e8d1f70..c7eabe111f 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 @@ async fn list_threads(mcp: &mut TestAppServer) -> Result { let list_id = mcp .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(50), 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 0a4ad00dab..17afc38480 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_list.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_list.rs @@ -103,6 +103,7 @@ async fn list_threads_with_sort( request_id, params: codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor, limit, @@ -144,6 +145,7 @@ async fn list_threads_for_relation( request_id, params: codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor, limit: Some(limit), @@ -776,6 +778,7 @@ async fn thread_list_respects_cwd_filters() -> Result<()> { let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -891,6 +894,7 @@ sqlite = true let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -1297,6 +1301,7 @@ sqlite = true let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -1335,6 +1340,7 @@ sqlite = true let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -1366,6 +1372,7 @@ sqlite = true let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -1553,6 +1560,7 @@ async fn thread_list_relation_filters_reject_invalid_requests() -> Result<()> { let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -1581,6 +1589,7 @@ async fn thread_list_relation_filters_reject_invalid_requests() -> Result<()> { let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -1873,6 +1882,7 @@ async fn thread_list_reports_loaded_subagent_direct_input_capability() -> Result request_id, params: codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -2696,6 +2706,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 { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(1), @@ -2739,6 +2750,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 { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: Some(backwards_cursor), limit: Some(10), @@ -3172,6 +3184,7 @@ async fn thread_list_invalid_cursor_returns_error() -> Result<()> { let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: Some("not-a-cursor".to_string()), limit: Some(2), 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 1a6437b63d..400dfa5ea6 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 @@ -89,6 +89,7 @@ async fn thread_section_move_pins_and_archives_before_first_turn() -> Result<()> let list_params = ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(100), @@ -294,6 +295,7 @@ async fn thread_section_move_pins_and_unpins_with_filtered_recency_pagination() let list_params = ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(1), @@ -487,6 +489,7 @@ async fn thread_sections_preserve_server_owned_manual_order_across_moves_and_res let list_params = ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), 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 aac4fc1151..c27ca5a0bd 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -407,6 +407,7 @@ async fn paginated_stored_thread_routes_projected_turns() -> Result<()> { let list_id = mcp .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(50), @@ -627,6 +628,7 @@ async fn thread_search_occurrences_reads_paginated_projection() -> Result<()> { ); store .create_thread(CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), @@ -1138,6 +1140,7 @@ async fn thread_list_includes_store_thread_without_rollout_path() -> Result<()> request_id: RequestId::Integer(1), params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), @@ -1520,6 +1523,7 @@ async fn paginated_thread_name_set_is_reflected_in_read_list_and_metadata_resume let list_id = mcp .send_thread_list_request(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(50), @@ -1726,6 +1730,7 @@ async fn paginated_history_lists_and_legacy_reads_use_projected_turns_and_items( ); store .create_thread(CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), @@ -2631,6 +2636,7 @@ async fn seed_pathless_store_thread( ) -> Result<()> { store .create_thread(CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 09b16cf4e6..19cd962b1b 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -4642,6 +4642,7 @@ async fn thread_resume_prefers_persisted_git_metadata_for_local_threads() -> Res let rollout_dir = rollout_path.parent().expect("rollout parent directory"); std::fs::create_dir_all(rollout_dir)?; let session_meta = SessionMeta { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: conversation_id.into(), diff --git a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs b/codex-rs/app-server/tests/suite/v2/thread_timeline.rs index 4dd2a6f77c..46a2a0b1bd 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_timeline.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_timeline.rs @@ -59,6 +59,7 @@ async fn timeline_pages_mix_items_and_resolve_the_opening_realtime_session() -> ); store .create_thread(CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs index 50041ea975..2a18e911fd 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs @@ -247,6 +247,7 @@ async fn thread_unarchive_preserves_pathless_store_metadata() -> Result<()> { let parent_thread_id = ThreadId::from_string("00000000-0000-4000-8000-000000000127")?; store .create_thread(CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), 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 a6e9595df2..822643bc7e 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -4599,6 +4599,7 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected( request_id, params: codex_app_server_protocol::ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(10), diff --git a/codex-rs/cli/tests/queue.rs b/codex-rs/cli/tests/queue.rs index a10615d86c..25aafe2e82 100644 --- a/codex-rs/cli/tests/queue.rs +++ b/codex-rs/cli/tests/queue.rs @@ -406,6 +406,7 @@ async fn remote_session_commands_with_workload_identity_use_server_auth() -> Res request_id: RequestId::Integer(4), params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, cursor: None, limit: None, sort_key: None, diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index d6742fc218..d4227af4b1 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -103,6 +103,7 @@ pub(crate) async fn run_codex_thread_interactive( let mut thread_extension_init = codex_extension_api::ExtensionDataInit::default(); thread_extension_init.insert(isolation); let spawn_result = Session::spawn(SessionSpawnArgs { + attached_thread: false, startup: Some(Arc::clone(&startup)), config, prompt_cache_key: parent_session diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 1824f43aad..073e7a19ee 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -485,6 +485,7 @@ impl From for CompactedHistoryInstallation<'_> { } pub(crate) struct SessionSpawnArgs { + pub(crate) attached_thread: bool, pub(crate) startup: Option>, pub(crate) config: Config, pub(crate) prompt_cache_key: PromptCacheKey, @@ -605,6 +606,7 @@ impl Session { .ensure_config_server_applied()?; args.config.validate_windows_mxc_requirement()?; let SessionSpawnArgs { + attached_thread, startup, config, prompt_cache_key, @@ -972,6 +974,7 @@ impl Session { let session = Box::pin(Session::new( startup.clone(), session_configuration, + attached_thread, prompt_cache_key, &environment_selections, config.clone(), diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 9d213c763d..9c415e6020 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -812,6 +812,7 @@ impl Session { pub(crate) async fn new( startup: Option>, mut session_configuration: SessionConfiguration, + attached_thread: bool, prompt_cache_key: PromptCacheKey, environment_selections: &[TurnEnvironmentSelection], config: Arc, @@ -1087,6 +1088,7 @@ impl Session { InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_) => { let auth = persistence_auth.await; let params = CreateThreadParams { + attached_thread, creator_user_id: auth.as_ref().and_then(CodexAuth::get_chatgpt_user_id), creator_account_id: auth.as_ref().and_then(CodexAuth::get_account_id), session_id, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 7ebdd51de8..182e03e588 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -4933,6 +4933,7 @@ async fn open_thread_persistence(session: &mut Session) -> PathBuf { let live_thread = LiveThread::create( Arc::clone(&session.services.thread_store), CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: session.session_id(), @@ -7186,6 +7187,7 @@ async fn make_session_with_config_and_rx( let session = Session::new( /*startup*/ None, session_configuration, + /*attached_thread*/ false, PromptCacheKey::default(), &default_environments, Arc::clone(&config), @@ -7323,6 +7325,7 @@ async fn make_session_with_history_source_and_agent_control_and_rx( let session = Session::new( /*startup*/ None, session_configuration, + /*attached_thread*/ false, PromptCacheKey::default(), &default_environments, Arc::clone(&config), @@ -8682,6 +8685,7 @@ async fn shutdown_complete_does_not_append_to_thread_store_after_shutdown() { let live_thread = LiveThread::create( Arc::clone(&thread_store), CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: session.session_id(), @@ -8802,6 +8806,7 @@ async fn submission_loop_channel_close_runs_full_thread_teardown() { let live_thread = LiveThread::create( Arc::clone(&thread_store), CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: session.session_id(), @@ -11736,6 +11741,7 @@ async fn attach_in_memory_thread_store( let live_thread = LiveThread::create( Arc::clone(&thread_store), CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: session.session_id(), diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index 142402b345..669d529856 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -1293,6 +1293,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { let mut thread_extension_init = codex_extension_api::ExtensionDataInit::default(); thread_extension_init.insert(codex_extension_api::SessionIsolation::Isolated); let (session, io) = Session::spawn(SessionSpawnArgs { + attached_thread: false, startup: None, config, prompt_cache_key: PromptCacheKey::default(), diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 10ab88e044..76ffa26d79 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -306,6 +306,8 @@ pub struct StartThreadOptions { pub history_mode: Option, pub session_source: Option, pub thread_source: Option, + /// Creation metadata forwarded to storage; never inherited from a fork source. + pub attached_thread: bool, pub dynamic_tools: Vec, pub metrics_service_name: Option, pub parent_trace: Option, @@ -337,6 +339,7 @@ impl StartThreadOptions { history_mode: None, session_source: None, thread_source: None, + attached_thread: false, dynamic_tools: Vec::new(), metrics_service_name: None, parent_trace: None, @@ -2175,6 +2178,7 @@ impl ThreadManagerState { history_mode, session_source, thread_source, + attached_thread, dynamic_tools, metrics_service_name, parent_trace, @@ -2397,6 +2401,7 @@ impl ThreadManagerState { let attachment_source = forked_from_thread_id.filter(|_| matches!(&initial_history, InitialHistory::Forked(_))); let spawn_result = Session::spawn(SessionSpawnArgs { + attached_thread, startup, config, prompt_cache_key, diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index 25ebd9989a..db37580ee3 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -557,6 +557,7 @@ async fn backfill_scans_existing_rollouts() -> Result<()> { fs::create_dir_all(parent).expect("should create rollout directory"); let session_meta_line = SessionMetaLine { meta: SessionMeta { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index a5b9eb2e68..db35aaa6bb 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -1872,6 +1872,7 @@ async fn resolve_resume_thread_id( request_id: RequestId::Integer(0), params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor, limit: Some(100), @@ -1959,6 +1960,7 @@ async fn resolve_resume_thread_id( request_id: RequestId::Integer(0), params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor, limit: Some(100), diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 141b4a72f4..b2cd4a7cd8 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -3285,6 +3285,9 @@ pub struct SessionMeta { pub id: ThreadId, #[serde(skip_serializing_if = "Option::is_none")] pub forked_from_id: Option, + /// Marks a persisted fork as attached to `forked_from_id`. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub attached_thread: bool, /// Exclusive ordinal inherited from the logical fork parent, independent of `history_base`. /// Revert may replace the physical history base while retaining this fork boundary. #[serde(default, skip_serializing_if = "Option::is_none")] @@ -3359,6 +3362,7 @@ impl Default for SessionMeta { session_id: id.into(), id, forked_from_id: None, + attached_thread: false, forked_from_ordinal_exclusive: None, parent_thread_id: None, timestamp: String::new(), diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs index 806039d624..59143ce39a 100644 --- a/codex-rs/rollout/src/compression_tests.rs +++ b/codex-rs/rollout/src/compression_tests.rs @@ -1121,6 +1121,7 @@ fn write_rollout(path: &std::path::Path, thread_id: ThreadId, message: &str) -> fs::create_dir_all(parent)?; let session_meta_line = SessionMetaLine { meta: SessionMeta { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index 66607d92ba..4742dd864a 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -122,6 +122,7 @@ pub type ConversationsPage = ThreadsPage; struct HeadTailSummary { originator: Option, saw_session_meta: bool, + attached_to_parent: bool, thread_id: Option, first_user_message: Option, preview: Option, @@ -290,6 +291,7 @@ impl<'a> RolloutFileVisitor for FilesByCreatedAtVisitor<'a> { self.cwd_filters, updated_at, PreviewFilter::RequirePreview, + /*include_attached_threads*/ false, ) .await { @@ -584,6 +586,7 @@ async fn traverse_directories_for_paths_updated( cwd_filters, updated_at_fallback, PreviewFilter::RequirePreview, + /*include_attached_threads*/ false, ) .await { @@ -642,6 +645,7 @@ async fn traverse_flat_paths_created( cwd_filters, updated_at, PreviewFilter::IncludeEmpty, + /*include_attached_threads*/ false, ) .await { @@ -704,6 +708,7 @@ async fn traverse_flat_paths_updated( cwd_filters, updated_at_fallback, PreviewFilter::IncludeEmpty, + /*include_attached_threads*/ false, ) .await { @@ -785,12 +790,16 @@ async fn build_thread_item( cwd_filters: Option<&[PathBuf]>, updated_at: Option, preview_filter: PreviewFilter, + include_attached_threads: bool, ) -> Option { // Read head and detect preview-bearing events; goal previews can appear before // the first normal user message. let summary = read_head_summary(&path, HEAD_RECORD_LIMIT) .await .unwrap_or_default(); + if summary.attached_to_parent && !include_attached_threads { + return None; + } if !allowed_sources.is_empty() && !summary .source @@ -882,6 +891,7 @@ pub async fn read_thread_item_from_rollout(path: PathBuf) -> Option /*cwd_filters*/ None, /*updated_at*/ None, PreviewFilter::RequirePreview, + /*include_attached_threads*/ true, ) .await } @@ -1004,6 +1014,7 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result, forked_from_id: Option, + attached_thread: bool, forked_from_ordinal_exclusive: Option, parent_thread_id: Option, source: Box, @@ -196,6 +197,17 @@ fn clone_io_error(err: &IoError) -> IoError { } impl RolloutRecorderParams { + pub fn with_attached_thread(mut self, attached_thread: bool) -> Self { + if let Self::Create { + attached_thread: value, + .. + } = &mut self + { + *value = attached_thread; + } + self + } + #[allow(clippy::too_many_arguments)] pub fn new( conversation_id: ThreadId, @@ -212,6 +224,7 @@ impl RolloutRecorderParams { conversation_id, rollout_id_override: None, forked_from_id, + attached_thread: false, forked_from_ordinal_exclusive: None, parent_thread_id, source: Box::new(source), @@ -918,6 +931,7 @@ impl RolloutRecorder { conversation_id, rollout_id_override, forked_from_id, + attached_thread, forked_from_ordinal_exclusive, parent_thread_id, source, @@ -952,6 +966,7 @@ impl RolloutRecorder { session_id, id: conversation_id, forked_from_id, + attached_thread, forked_from_ordinal_exclusive: forked_from_ordinal_exclusive .filter(|_| forked_from_id.is_some()), parent_thread_id, diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index d42a7c4ba7..a2843aada8 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -214,6 +214,7 @@ async fn state_db_init_backfills_before_returning() -> anyhow::Result<()> { let session_meta_line = SessionMetaLine { meta: SessionMeta { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), @@ -1272,13 +1273,58 @@ async fn append_rollout_item_to_path_assigns_next_paginated_ordinal() -> std::io } #[tokio::test] -async fn list_threads_db_disabled_does_not_skip_paginated_items() -> std::io::Result<()> { +async fn discovery_without_database_skips_attached_threads_before_pagination() -> std::io::Result<()> +{ let home = TempDir::new().expect("temp dir"); let config = test_config(home.path()); let newest = write_session_file(home.path(), "2025-01-03T12-00-00", Uuid::from_u128(9001))?; let middle = write_session_file(home.path(), "2025-01-02T12-00-00", Uuid::from_u128(9002))?; let _oldest = write_session_file(home.path(), "2025-01-01T12-00-00", Uuid::from_u128(9003))?; + let parent = ThreadId::from_string(&Uuid::from_u128(9001).to_string()).unwrap(); + crate::append_thread_name(home.path(), parent, "shared title").await?; + for (timestamp, uuid, mode) in [ + ( + "2025-01-04T12-00-00", + Uuid::from_u128(9004), + ThreadHistoryMode::Legacy, + ), + ( + "2025-01-03T06-00-00", + Uuid::from_u128(9005), + ThreadHistoryMode::Paginated, + ), + ] { + let path = write_session_file(home.path(), timestamp, uuid)?; + let mut lines = read_rollout_lines(&path)?; + let RolloutItem::SessionMeta(meta) = &mut lines[0].item else { + panic!("fixture must start with session metadata"); + }; + meta.meta.attached_thread = true; + meta.meta.forked_from_id = Some(parent); + meta.meta.history_mode = mode; + let mut file = File::create(&path)?; + for line in lines { + writeln!(file, "{}", serde_json::to_string(&line)?)?; + } + let child = ThreadId::from_string(&uuid.to_string()).unwrap(); + crate::append_thread_name(home.path(), child, "shared title").await?; + // Hiding discovery must not prevent opening an explicitly selected attached thread. + assert_eq!( + crate::read_thread_item_from_rollout(path) + .await + .unwrap() + .thread_id, + Some(child) + ); + } + let named = crate::session_index::find_thread_meta_by_name_str( + home.path(), + "shared title", + /*state_db_ctx*/ None, + ) + .await?; + assert_eq!(named.map(|(_, meta)| meta.meta.id), Some(parent)); let default_provider = config.model_provider_id.clone(); let page1 = RolloutRecorder::list_threads( diff --git a/codex-rs/rollout/src/rollout_reference_index.rs b/codex-rs/rollout/src/rollout_reference_index.rs index 2b21b82487..107dd62325 100644 --- a/codex-rs/rollout/src/rollout_reference_index.rs +++ b/codex-rs/rollout/src/rollout_reference_index.rs @@ -1,4 +1,4 @@ -//! Indexes direct fork references found in local rollout files. +//! Indexes direct fork references and attached children found in local rollout files. use std::collections::HashMap; use std::collections::HashSet; @@ -16,10 +16,10 @@ use crate::SESSIONS_SUBDIR; use crate::compression::RolloutFile; use crate::rollout_file_name::RolloutFileName; -/// Direct history-base edges discovered from local rollout metadata. +/// Fork relationships discovered from local rollout metadata. /// -/// This indexes immutable rollout IDs, not a thread's selected lineage. Callers use it to answer -/// cheap inverse-reference questions without each reimplementing rollout discovery. +/// Rollouts are indexed by immutable rollout IDs, not a thread's selected lineage. Attached +/// children refer to their parent's stable thread ID. #[derive(Debug, Default)] pub struct RolloutReferenceIndex { rollouts_by_id: HashMap, @@ -31,6 +31,7 @@ struct IndexedRollout { thread_id: ThreadId, path: PathBuf, history_base: Option, + attached_to_thread_id: Option, } impl RolloutReferenceIndex { @@ -129,6 +130,10 @@ impl RolloutReferenceIndex { thread_id: meta.meta.id, path: rollout_file.into_path(), history_base: meta.meta.history_base, + attached_to_thread_id: meta + .meta + .forked_from_id + .filter(|_| meta.meta.attached_thread), }); } } @@ -167,6 +172,17 @@ impl RolloutReferenceIndex { .and_then(|rollout| rollout.history_base.as_ref()) } + /// Find an indexed attached child of any candidate parent. + pub fn attached_child(&self, parent_thread_ids: &[ThreadId]) -> Option { + let parents: HashSet<_> = parent_thread_ids.iter().copied().collect(); + self.rollouts_by_id.values().find_map(|rollout| { + rollout + .attached_to_thread_id + .filter(|parent| parents.contains(parent)) + .map(|_| rollout.thread_id) + }) + } + /// Returns rollout IDs and paths whose session metadata belongs to `thread_id`. pub fn rollouts_for_thread( &self, diff --git a/codex-rs/rollout/src/session_index.rs b/codex-rs/rollout/src/session_index.rs index b69b388ed1..6d504c9eb3 100644 --- a/codex-rs/rollout/src/session_index.rs +++ b/codex-rs/rollout/src/session_index.rs @@ -224,6 +224,7 @@ pub async fn find_thread_meta_candidates_by_name_str( ) .await? && let Ok(session_meta) = super::list::read_session_meta_line(&path).await + && !session_meta.meta.attached_thread && (allowed_sources.is_empty() || allowed_sources.contains(&session_meta.meta.source)) && session_meta .meta diff --git a/codex-rs/rollout/src/session_index_tests.rs b/codex-rs/rollout/src/session_index_tests.rs index 32eb6c1a07..7f8dd4d527 100644 --- a/codex-rs/rollout/src/session_index_tests.rs +++ b/codex-rs/rollout/src/session_index_tests.rs @@ -35,6 +35,7 @@ fn write_rollout_with_source_and_provider( ordinal: None, item: RolloutItem::SessionMeta(Box::new(SessionMetaLine { meta: SessionMeta { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/rollout/src/state_db_tests.rs b/codex-rs/rollout/src/state_db_tests.rs index ebf5881870..b2afea4407 100644 --- a/codex-rs/rollout/src/state_db_tests.rs +++ b/codex-rs/rollout/src/state_db_tests.rs @@ -325,6 +325,7 @@ fn write_rollout_with_user_message( ordinal: None, item: RolloutItem::SessionMeta(Box::new(SessionMetaLine { meta: SessionMeta { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/rollout/src/tests.rs b/codex-rs/rollout/src/tests.rs index b1484f1694..4db5a849b9 100644 --- a/codex-rs/rollout/src/tests.rs +++ b/codex-rs/rollout/src/tests.rs @@ -1593,6 +1593,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> { ordinal: None, item: RolloutItem::SessionMeta(Box::new(SessionMetaLine { meta: SessionMeta { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: conversation_id.into(), diff --git a/codex-rs/state/migrations/0062_threads_attached_parent.sql b/codex-rs/state/migrations/0062_threads_attached_parent.sql new file mode 100644 index 0000000000..8cb4fe4c56 --- /dev/null +++ b/codex-rs/state/migrations/0062_threads_attached_parent.sql @@ -0,0 +1,4 @@ +ALTER TABLE threads ADD COLUMN attached_thread_parent_id TEXT; + +CREATE INDEX idx_threads_attached_parent ON threads(attached_thread_parent_id) + WHERE attached_thread_parent_id IS NOT NULL; diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index 18bcec64cd..a501301553 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -94,6 +94,10 @@ fn apply_session_meta_from_item(metadata: &mut ThreadMetadata, meta_line: &Sessi .take() .or_else(|| meta_line.meta.creator_account_id.clone()); metadata.id = meta_line.meta.id; + metadata.attached_thread_parent_id = meta_line + .meta + .forked_from_id + .filter(|_| meta_line.meta.attached_thread); metadata.source = enum_to_string(&meta_line.meta.source); if metadata.originator.is_none() && !meta_line.meta.originator.is_empty() { metadata.originator = Some(meta_line.meta.originator.clone()); @@ -528,6 +532,7 @@ mod tests { &RolloutItem::SessionMeta(Box::new(SessionMetaLine { meta: SessionMeta { creator_user_id: None, + attached_thread: false, creator_account_id: None, session_id: thread_id.into(), id: thread_id, @@ -778,6 +783,7 @@ mod tests { &RolloutItem::SessionMeta(Box::new(SessionMetaLine { meta: SessionMeta { creator_user_id: None, + attached_thread: false, creator_account_id: None, session_id: thread_id.into(), id: thread_id, @@ -820,6 +826,7 @@ mod tests { let created_at = DateTime::::from_timestamp(1_735_689_600, 0).expect("timestamp"); ThreadMetadata { creator_user_id: None, + attached_thread_parent_id: None, creator_account_id: None, originator: None, id, diff --git a/codex-rs/state/src/model/thread_metadata.rs b/codex-rs/state/src/model/thread_metadata.rs index 9237a57324..223519f50b 100644 --- a/codex-rs/state/src/model/thread_metadata.rs +++ b/codex-rs/state/src/model/thread_metadata.rs @@ -36,13 +36,15 @@ pub enum SortDirection { Desc, } -/// Spawn-graph relationship used to filter thread listings. +/// Thread relationship used to filter thread listings. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ThreadRelationFilter { /// Return only threads whose immediate parent is the given thread. DirectChildrenOf(ThreadId), /// Return every thread transitively descended from the given thread. DescendantsOf(ThreadId), + /// Return only saved threads attached to this parent. + AttachedThreadsOf(ThreadId), } /// A pagination anchor used for keyset pagination. @@ -123,6 +125,8 @@ pub struct ExtractionOutcome { /// Canonical persisted thread metadata. #[derive(Debug, Clone, PartialEq, Eq)] pub struct ThreadMetadata { + /// The parent of an attached thread, derived from canonical rollout metadata. + pub attached_thread_parent_id: Option, /// Originator recorded at creation, if available. pub originator: Option, /// ChatGPT user that created the thread, if known. @@ -297,6 +301,7 @@ impl ThreadMetadataBuilder { .unwrap_or(updated_at); let guardian_review = crate::is_guardian_review_source(&self.source); ThreadMetadata { + attached_thread_parent_id: None, originator: self.originator.clone(), creator_user_id: self.creator_user_id.clone(), creator_account_id: self.creator_account_id.clone(), @@ -501,6 +506,7 @@ fn canonicalize_datetime(dt: DateTime) -> DateTime { #[derive(Debug)] pub(crate) struct ThreadRow { + attached_thread_parent_id: Option, originator: Option, creator_user_id: Option, creator_account_id: Option, @@ -543,6 +549,7 @@ pub(crate) struct ThreadRow { impl ThreadRow { pub(crate) fn try_from_row(row: &SqliteRow) -> Result { Ok(Self { + attached_thread_parent_id: row.try_get("attached_thread_parent_id")?, originator: row.try_get("originator")?, creator_user_id: row.try_get("creator_user_id")?, creator_account_id: row.try_get("creator_account_id")?, @@ -589,6 +596,7 @@ impl TryFrom for ThreadMetadata { fn try_from(row: ThreadRow) -> std::result::Result { let ThreadRow { + attached_thread_parent_id, originator, creator_user_id, creator_account_id, @@ -650,6 +658,9 @@ impl TryFrom for ThreadMetadata { }; Ok(Self { id: ThreadId::try_from(id)?, + attached_thread_parent_id: attached_thread_parent_id + .map(ThreadId::try_from) + .transpose()?, originator, creator_user_id, creator_account_id, @@ -766,6 +777,7 @@ mod tests { fn thread_row(reasoning_effort: Option<&str>) -> ThreadRow { ThreadRow { + attached_thread_parent_id: None, originator: None, creator_user_id: None, creator_account_id: None, @@ -808,6 +820,7 @@ mod tests { fn expected_thread_metadata(reasoning_effort: Option) -> ThreadMetadata { ThreadMetadata { + attached_thread_parent_id: None, originator: None, creator_user_id: None, creator_account_id: None, diff --git a/codex-rs/state/src/runtime/memories.rs b/codex-rs/state/src/runtime/memories.rs index f11a779fb6..18c32ea49b 100644 --- a/codex-rs/state/src/runtime/memories.rs +++ b/codex-rs/state/src/runtime/memories.rs @@ -177,6 +177,7 @@ WHERE kind = ? AND job_key = ? r#" SELECT threads.id, + threads.attached_thread_parent_id, threads.rollout_path, threads.created_at_ms AS created_at, threads.updated_at_ms AS updated_at, @@ -571,6 +572,7 @@ WHERE so.thread_id = ? AND so.source_updated_at = ? r#" SELECT threads.id, + threads.attached_thread_parent_id, threads.rollout_path, threads.created_at_ms AS created_at, threads.updated_at_ms AS updated_at, @@ -2089,6 +2091,7 @@ mod tests { ); eligible_idle.created_at = eligible_idle_at; eligible_idle.updated_at = eligible_idle_at; + eligible_idle.attached_thread_parent_id = Some(current_thread_id); runtime .upsert_thread(&eligible_idle) .await diff --git a/codex-rs/state/src/runtime/test_support.rs b/codex-rs/state/src/runtime/test_support.rs index a259dd93b8..5d2f7fe85b 100644 --- a/codex-rs/state/src/runtime/test_support.rs +++ b/codex-rs/state/src/runtime/test_support.rs @@ -46,6 +46,7 @@ pub(super) fn test_thread_metadata( let now = DateTime::::from_timestamp(1_700_000_000, 0).expect("timestamp"); ThreadMetadata { creator_user_id: None, + attached_thread_parent_id: None, creator_account_id: None, originator: None, id: thread_id, diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 5a96d13c58..c72a9222f7 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -7,12 +7,15 @@ use codex_protocol::protocol::SessionSource; use std::sync::atomic::AtomicI64; use std::sync::atomic::Ordering; +const EXCLUDE_ATTACHED_THREADS: &str = " AND threads.attached_thread_parent_id IS NULL"; + impl StateRuntime { pub async fn get_thread(&self, id: ThreadId) -> anyhow::Result> { let row = sqlx::query( r#" SELECT threads.id, + threads.attached_thread_parent_id, threads.rollout_path, threads.created_at_ms AS created_at, threads.updated_at_ms AS updated_at, @@ -454,6 +457,7 @@ ON CONFLICT(child_thread_id) DO NOTHING }, /*include_thread_id_tiebreaker*/ false, ); + builder.push(EXCLUDE_ATTACHED_THREADS); builder.push(" AND threads.title = "); builder.push_bind(title); if let Some(cwd) = cwd { @@ -585,6 +589,7 @@ ON CONFLICT(child_thread_id) DO NOTHING | crate::SortKey::SectionPosition ), ); + builder.push(EXCLUDE_ATTACHED_THREADS); push_thread_order_and_limit( &mut builder, sort_key, @@ -663,8 +668,9 @@ INSERT INTO threads ( git_origin_url, memory_mode, project_id, - daybreak_enabled -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + daybreak_enabled, + attached_thread_parent_id +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO NOTHING "#, ) @@ -718,6 +724,7 @@ ON CONFLICT(id) DO NOTHING .bind("enabled") .bind(metadata.project_id.as_deref()) .bind(metadata.daybreak_enabled) + .bind(metadata.attached_thread_parent_id.map(|id| id.to_string())) .execute_with_busy_retry(self.pool.as_ref()) .await?; self.insert_thread_spawn_edge_from_source_if_absent(metadata.id, metadata.source.as_str()) @@ -963,9 +970,11 @@ INSERT INTO threads ( git_origin_url, memory_mode, project_id, - daybreak_enabled -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + daybreak_enabled, + attached_thread_parent_id +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET + attached_thread_parent_id = COALESCE(excluded.attached_thread_parent_id, threads.attached_thread_parent_id), rollout_path = excluded.rollout_path, created_at = excluded.created_at, updated_at = excluded.updated_at, @@ -1054,6 +1063,7 @@ ON CONFLICT(id) DO UPDATE SET .bind(creation_memory_mode.unwrap_or("enabled")) .bind(metadata.project_id.as_deref()) .bind(metadata.daybreak_enabled) + .bind(metadata.attached_thread_parent_id.map(|id| id.to_string())) .execute_with_busy_retry(self.pool.as_ref()) .await?; self.insert_thread_spawn_edge_from_source_if_absent(metadata.id, metadata.source.as_str()) @@ -1272,6 +1282,9 @@ 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(_)) => { + builder.push(", NULL AS parent_thread_id\nFROM threads") + } None => builder.push(" FROM threads"), }; let include_thread_id_tiebreaker = relation_filter.is_some() @@ -1285,6 +1298,12 @@ WITH RECURSIVE subtree(child_thread_id, parent_thread_id) AS ( include_thread_id_tiebreaker, /*include_empty_preview*/ relation_filter.is_some(), ); + if !matches!( + relation_filter, + Some(crate::ThreadRelationFilter::AttachedThreadsOf(_)) + ) { + builder.push(EXCLUDE_ATTACHED_THREADS); + } match relation_filter { Some(crate::ThreadRelationFilter::DirectChildrenOf(parent_thread_id)) => { builder.push(" AND listed_edge.parent_thread_id = "); @@ -1294,11 +1313,15 @@ WITH RECURSIVE subtree(child_thread_id, parent_thread_id) AS ( builder.push(" AND subtree.child_thread_id != "); builder.push_bind(ancestor_thread_id.to_string()); } + Some(crate::ThreadRelationFilter::AttachedThreadsOf(parent)) => { + builder.push(" AND threads.attached_thread_parent_id = "); + builder.push_bind(parent.to_string()); + } 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 spawn-edge index and sort the matching subtree instead of scanning the + // 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, // Multi-cwd listing is supported but at the time of writing has no current use in production. @@ -1321,6 +1344,7 @@ pub(super) fn push_thread_select_columns(builder: &mut QueryBuilder) { r#" SELECT threads.id, + threads.attached_thread_parent_id, threads.rollout_path, threads.created_at_ms AS created_at, threads.updated_at_ms AS updated_at, @@ -1708,6 +1732,81 @@ mod tests { ); } + #[tokio::test] + async fn attached_thread_discovery_preserves_parent_lifecycle_and_cwd() -> anyhow::Result<()> { + let home = unique_temp_dir(); + let config = crate::SqliteConfig::new_for_testing(home.as_path().abs()); + let runtime = StateRuntime::init(config, "test-provider".to_string()).await?; + let parent = ThreadId::new(); + let child = ThreadId::new(); + let mut parent_metadata = test_thread_metadata(&home, parent, home.clone()); + parent_metadata.preview = Some("parent".into()); + parent_metadata.title = "shared title".into(); + let mut child_metadata = test_thread_metadata(&home, child, home.join("original")); + child_metadata.attached_thread_parent_id = Some(parent); + child_metadata.preview = Some("saved child".into()); + child_metadata.created_at = Utc::now() + chrono::Duration::seconds(5); + child_metadata.updated_at = child_metadata.created_at; + child_metadata.title = parent_metadata.title.clone(); + runtime.upsert_thread(&parent_metadata).await?; + assert!(runtime.insert_thread_if_absent(&child_metadata).await?); + let filters = ThreadFilterOptions { + archived_only: false, + allowed_sources: &[], + model_providers: None, + cwd_filters: None, + section: None, + project_id: None, + anchor: None, + sort_key: SortKey::CreatedAt, + sort_direction: SortDirection::Desc, + search_term: None, + }; + let named = runtime + .find_thread_by_exact_title( + "shared title", + &[], + /*model_providers*/ None, + /*archived_only*/ false, + /*cwd*/ None, + ) + .await?; + assert_eq!(named.map(|thread| thread.id), Some(parent)); + let ids = runtime + .list_thread_ids( + /*limit*/ 1, + /*anchor*/ None, + SortKey::UpdatedAt, + &[], + /*model_providers*/ None, + /*archived_only*/ false, + ) + .await?; + assert_eq!(ids.len(), 1); + assert_ne!(ids[0], child); + parent_metadata.cwd = home.join("moved"); + runtime.upsert_thread(&parent_metadata).await?; + runtime + .mark_archived(parent, &parent_metadata.rollout_path, Utc::now()) + .await?; + let saved = runtime + .list_threads_by_relation( + /*page_size*/ 10, + crate::ThreadRelationFilter::AttachedThreadsOf(parent), + filters, + ) + .await?; + assert_eq!( + saved + .items + .iter() + .map(|t| (t.id, t.cwd.clone())) + .collect::>(), + vec![(child, child_metadata.cwd)] + ); + Ok(()) + } + #[tokio::test] async fn archived_threads_without_previews_remain_listed() { let codex_home = unique_temp_dir(); @@ -2808,6 +2907,7 @@ mod tests { ); let items = vec![RolloutItem::SessionMeta(Box::new(SessionMetaLine { meta: SessionMeta { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), @@ -2882,6 +2982,7 @@ mod tests { ); let items = vec![RolloutItem::SessionMeta(Box::new(SessionMetaLine { meta: SessionMeta { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 14a967e6e4..2df41dc67e 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -166,6 +166,7 @@ mod tests { ] { store .create_thread(CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), @@ -442,6 +443,7 @@ mod tests { history_mode: ThreadHistoryMode, ) -> CreateThreadParams { CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), @@ -583,6 +585,7 @@ impl InMemoryThreadStore { let mut state = self.state.lock().await; state.calls.create_thread += 1; let session_meta = SessionMeta { + attached_thread: params.attached_thread, session_id: params.session_id, id: params.thread_id, forked_from_id: params.forked_from_id, @@ -1002,6 +1005,11 @@ 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::DescendantsOf(ancestor_thread_id)) => { let mut subtree = HashSet::from([ancestor_thread_id]); loop { diff --git a/codex-rs/thread-store/src/local/create_thread.rs b/codex-rs/thread-store/src/local/create_thread.rs index e3b7fc8a89..8b9823eb4d 100644 --- a/codex-rs/thread-store/src/local/create_thread.rs +++ b/codex-rs/thread-store/src/local/create_thread.rs @@ -41,6 +41,7 @@ pub(super) async fn create_thread( ) .with_creator(params.creator_user_id, params.creator_account_id) .with_session_id(params.session_id) + .with_attached_thread(params.attached_thread) .with_selected_capability_roots(params.selected_capability_roots) .with_runtime_workspace_roots(params.runtime_workspace_roots.map(|roots| { roots diff --git a/codex-rs/thread-store/src/local/delete_thread.rs b/codex-rs/thread-store/src/local/delete_thread.rs index 75e55ef84f..ab13996c74 100644 --- a/codex-rs/thread-store/src/local/delete_thread.rs +++ b/codex-rs/thread-store/src/local/delete_thread.rs @@ -12,6 +12,7 @@ use std::io::ErrorKind; use std::path::Path; use std::path::PathBuf; +use codex_protocol::ThreadId; use codex_rollout::ARCHIVED_SESSIONS_SUBDIR; use codex_rollout::RolloutReferenceIndex; use codex_rollout::SESSIONS_SUBDIR; @@ -33,6 +34,32 @@ struct ThreadRollouts { paths: Vec, } +impl LocalThreadStore { + /// Reject deletion while any candidate has attached threads. Call before stopping live work; + /// deletion repeats this check under existing writer ownership to cover concurrent creation. + pub async fn ensure_no_attached_threads( + &self, + thread_ids: &[ThreadId], + ) -> ThreadStoreResult<()> { + let reference_index = scan_reference_index(self).await?; + check_attached_children(&reference_index, thread_ids) + } +} + +fn check_attached_children( + reference_index: &RolloutReferenceIndex, + thread_ids: &[ThreadId], +) -> ThreadStoreResult<()> { + let Some(child) = reference_index.attached_child(thread_ids) else { + return Ok(()); + }; + Err(ThreadStoreError::InvalidRequest { + message: format!( + "cannot delete thread while attached thread {child} exists; delete the attached thread first" + ), + }) +} + impl ThreadRollouts { fn from_index( reference_index: &RolloutReferenceIndex, @@ -70,10 +97,11 @@ pub(super) async fn delete_thread( let thread_id = params.thread_id; let _lifecycle_guard = store.live_writer_locks.lock_lifecycle(thread_id).await; let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; + let mut writer_guards = store.acquire_writer_locks(&[thread_id]).await?; let reference_index = scan_reference_index(store).await?; let thread_rollouts = ThreadRollouts::from_index(&reference_index, thread_id); + check_attached_children(&reference_index, &[thread_id])?; ensure_no_external_references(&reference_index, std::slice::from_ref(&thread_rollouts))?; - let mut writer_guards = store.acquire_writer_locks(&[thread_id]).await?; if let Some(cleanup) = &store.thread_data_cleanup { cleanup(vec![thread_id]).await?; } @@ -113,14 +141,14 @@ pub(super) async fn delete_threads( _live_writer_guards.push(store.live_writer_locks.lock(thread_id).await); } + let mut writer_guards = store.acquire_writer_locks(&lock_thread_ids).await?; let reference_index = scan_reference_index(store).await?; let thread_rollouts = thread_ids .iter() .map(|thread_id| ThreadRollouts::from_index(&reference_index, *thread_id)) .collect::>(); + check_attached_children(&reference_index, &lock_thread_ids)?; ensure_no_external_references(&reference_index, thread_rollouts.as_slice())?; - - let mut writer_guards = store.acquire_writer_locks(&lock_thread_ids).await?; if let Some(cleanup) = &store.thread_data_cleanup { cleanup(thread_ids.clone()).await?; } @@ -136,7 +164,7 @@ pub(super) async fn delete_threads( Ok(()) } -async fn delete_state_rows( +pub(super) async fn delete_state_rows( store: &LocalThreadStore, thread_ids: &[codex_protocol::ThreadId], ) -> ThreadStoreResult { @@ -272,7 +300,10 @@ async fn delete_thread_after_reference_check( Ok(()) } -fn delete_rollout_file(store: &LocalThreadStore, rollout_path: &Path) -> ThreadStoreResult { +pub(super) fn delete_rollout_file( + store: &LocalThreadStore, + rollout_path: &Path, +) -> ThreadStoreResult { let plain_path = codex_rollout::plain_rollout_path(rollout_path); let compressed_path = plain_path.with_extension("jsonl.zst"); let deleted_plain = delete_rollout_path(store, plain_path.as_path())?; diff --git a/codex-rs/thread-store/src/local/list_threads.rs b/codex-rs/thread-store/src/local/list_threads.rs index dc53964f98..fea544636d 100644 --- a/codex-rs/thread-store/src/local/list_threads.rs +++ b/codex-rs/thread-store/src/local/list_threads.rs @@ -198,23 +198,10 @@ async fn list_section_threads( }, search_term: params.search_term.as_deref(), }; - let page = match params.relation_filter { - Some(ThreadRelationFilter::DirectChildrenOf(thread_id)) => { + let page = match params.relation_filter.map(state_relation_filter) { + Some(relation_filter) => { state_db - .list_threads_by_relation( - params.page_size, - codex_state::ThreadRelationFilter::DirectChildrenOf(thread_id), - filters, - ) - .await - } - Some(ThreadRelationFilter::DescendantsOf(thread_id)) => { - state_db - .list_threads_by_relation( - params.page_size, - codex_state::ThreadRelationFilter::DescendantsOf(thread_id), - filters, - ) + .list_threads_by_relation(params.page_size, relation_filter, filters) .await } None => state_db.list_threads(params.page_size, filters).await, @@ -255,16 +242,6 @@ pub(super) async fn list_rollout_threads( sort_direction: codex_rollout::SortDirection, ) -> ThreadStoreResult { if params.relation_filter.is_some() || params.section.is_some() || params.project_id.is_some() { - let relation_filter = params - .relation_filter - .map(|relation_filter| match relation_filter { - ThreadRelationFilter::DirectChildrenOf(parent_thread_id) => { - codex_state::ThreadRelationFilter::DirectChildrenOf(parent_thread_id) - } - ThreadRelationFilter::DescendantsOf(ancestor_thread_id) => { - codex_state::ThreadRelationFilter::DescendantsOf(ancestor_thread_id) - } - }); let page = codex_rollout::state_db::list_threads_db( state_db.as_deref(), &config.sqlite, @@ -275,7 +252,7 @@ pub(super) async fn list_rollout_threads( params.allowed_sources.as_slice(), params.model_providers.as_deref(), params.cwd_filters.as_deref(), - relation_filter, + params.relation_filter.map(state_relation_filter), params.archived, params.section.as_ref().map(Option::as_deref), params.project_id.as_ref().map(Option::as_deref), @@ -354,6 +331,20 @@ pub(super) async fn list_rollout_threads( }) } +fn state_relation_filter(filter: ThreadRelationFilter) -> codex_state::ThreadRelationFilter { + match filter { + ThreadRelationFilter::DirectChildrenOf(id) => { + codex_state::ThreadRelationFilter::DirectChildrenOf(id) + } + ThreadRelationFilter::DescendantsOf(id) => { + codex_state::ThreadRelationFilter::DescendantsOf(id) + } + ThreadRelationFilter::AttachedThreadsOf(id) => { + codex_state::ThreadRelationFilter::AttachedThreadsOf(id) + } + } +} + #[cfg(test)] mod tests { use chrono::Utc; diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index a8213b8845..0fba3f55f8 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -28,11 +28,45 @@ pub(super) async fn create_thread( params: CreateThreadParams, ) -> ThreadStoreResult<()> { let thread_id = params.thread_id; + let attachment = if params.attached_thread { + let parent = params + .forked_from_id + .filter(|parent| *parent != thread_id) + .ok_or_else(|| ThreadStoreError::InvalidRequest { + message: "attached threads require a distinct fork source".to_string(), + })?; + let state = store + .state_db + .as_ref() + .ok_or(ThreadStoreError::Unsupported { + operation: "attached_threads", + })?; + // Keep parent deletion out until the child's canonical metadata is durable. + let source_ownership = ( + store.live_writer_locks.lock(parent).await, + store.acquire_writer_locks(&[parent]).await?, + ); + super::thread_rollout_resolver::resolve_current_including_archived(store, parent) + .await? + .ok_or(ThreadStoreError::ThreadNotFound { thread_id: parent })?; + Some((state, source_ownership)) + } else { + None + }; let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; let history_mode = params.history_mode; store.ensure_live_recorder_absent(thread_id).await?; let writer_lock = store.acquire_writer_lock(thread_id)?; let recorder = create_thread::create_thread(store, params, writer_lock.clone()).await?; + if let Some((state, _source_ownership)) = attachment { + recorder.persist().await.map_err(thread_store_io_error)?; + if let Err(error) = index_attached_thread(store, state, recorder.rollout_path()).await { + recorder.discard().await.map_err(thread_store_io_error)?; + super::delete_thread::delete_rollout_file(store, recorder.rollout_path())?; + super::delete_thread::delete_state_rows(store, &[thread_id]).await?; + return Err(error); + } + } store .insert_live_recorder(thread_id, recorder, thread_id, history_mode, writer_lock) .await @@ -235,6 +269,40 @@ pub(super) async fn discard_thread( .map_err(thread_store_io_error) } +async fn index_attached_thread( + store: &LocalThreadStore, + state: &codex_state::StateRuntime, + path: &std::path::Path, +) -> ThreadStoreResult<()> { + let meta = codex_rollout::read_session_meta_line(path) + .await + .map_err(thread_store_io_error)?; + let memory_mode = meta.meta.memory_mode.clone(); + let items = [RolloutItem::SessionMeta(Box::new(meta))]; + let mut builder = codex_rollout::builder_from_items(&items, path).ok_or_else(|| { + ThreadStoreError::Internal { + message: format!("invalid attached thread metadata at {}", path.display()), + } + })?; + if super::helpers::rollout_path_is_archived(&store.config.codex_home, path) { + builder.archived_at = Some(builder.updated_at.unwrap_or(builder.created_at)); + } + state + .apply_rollout_items( + &builder, + &items, + memory_mode.as_deref(), + /*updated_at_override*/ None, + ) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!( + "failed to index attached thread at {}: {err}", + path.display() + ), + }) +} + pub(super) async fn rollout_path( store: &LocalThreadStore, thread_id: ThreadId, diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 98c6ff70c2..4a7467a0df 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -1602,6 +1602,120 @@ mod tests { } } + #[tokio::test] + async fn attached_creation_and_deletion_coordinate_across_stores() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let runtime = codex_state::StateRuntime::init( + config.sqlite.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("initialize state db"); + let first = LocalThreadStore::new(config.clone(), Some(runtime.clone())); + let second = LocalThreadStore::new(config, Some(runtime.clone())); + let parent_uuid = uuid::Uuid::new_v4(); + let parent = ThreadId::from_u128(parent_uuid.as_u128()); + let parent_path = write_session_file(home.path(), "2025-01-03T00-00-00", parent_uuid) + .expect("write parent"); + let child = ThreadId::new(); + let mut params = create_thread_params(child); + params.attached_thread = true; + assert!(matches!( + second.create_thread(params.clone()).await, + Err(ThreadStoreError::InvalidRequest { .. }) + )); + params.forked_from_id = Some(parent); + let parent_guard = first.acquire_writer_lock(parent).expect("lock parent"); + assert!(matches!( + second.create_thread(params.clone()).await, + Err(ThreadStoreError::Conflict { .. }) + )); + drop(parent_guard); + let pool = runtime + .sqlite() + .open_read_write_pool(&runtime.sqlite().state_db_path()) + .await + .expect("open state db"); + sqlx::query( + "CREATE TRIGGER fail_attached_index BEFORE UPDATE OF memory_mode ON threads WHEN NEW.attached_thread_parent_id IS NOT NULL BEGIN SELECT RAISE(ABORT, 'index failed'); END", + ) + .execute(&pool) + .await + .expect("inject index failure"); + let mut failing_params = params.clone(); + failing_params.metadata.memory_mode = ThreadMemoryMode::Disabled; + second + .create_thread(failing_params) + .await + .expect_err("indexing must fail"); + first + .ensure_no_attached_threads(&[parent]) + .await + .expect("failed creation must not leave an attached rollout"); + assert!( + runtime + .get_thread(child) + .await + .expect("read index") + .is_none() + ); + sqlx::query("DROP TRIGGER fail_attached_index") + .execute(&pool) + .await + .expect("remove index failure"); + second + .create_thread(params.clone()) + .await + .expect("create child"); + assert_eq!( + runtime + .delete_thread(child) + .await + .expect("remove index row"), + 1 + ); + assert!(matches!( + first + .delete_threads(DeleteThreadsParams { + thread_ids: vec![parent] + }) + .await, + Err(ThreadStoreError::InvalidRequest { .. }) + )); + assert!( + parent_path.exists(), + "refused deletion must preserve parent" + ); + second + .discard_thread(child) + .await + .expect("release child writer"); + assert!(matches!( + first + .delete_threads(DeleteThreadsParams { + thread_ids: vec![parent] + }) + .await, + Err(ThreadStoreError::InvalidRequest { .. }) + )); + first + .delete_thread(DeleteThreadParams { thread_id: child }) + .await + .expect("delete child"); + first + .delete_threads(DeleteThreadsParams { + thread_ids: vec![parent], + }) + .await + .expect("parent can be deleted after its child"); + assert!(!parent_path.exists()); + assert!(matches!( + second.create_thread(params).await, + Err(ThreadStoreError::ThreadNotFound { thread_id }) if thread_id == parent + )); + } + #[tokio::test] async fn resume_thread_reopens_live_writer_and_appends_with_stale_sqlite_path() { let home = TempDir::new().expect("temp dir"); @@ -2246,6 +2360,7 @@ mod tests { fn create_thread_params(thread_id: ThreadId) -> CreateThreadParams { CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs b/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs index f9b4c63348..8c6856588c 100644 --- a/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs +++ b/codex-rs/thread-store/src/local/pending_thread_metadata_tests.rs @@ -457,6 +457,7 @@ fn staged_model_patch() -> ThreadMetadataPatch { fn create_thread_params(thread_id: ThreadId) -> CreateThreadParams { CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/thread-store/src/local/revert_thread.rs b/codex-rs/thread-store/src/local/revert_thread.rs index 424a0f54c1..8a9166365f 100644 --- a/codex-rs/thread-store/src/local/revert_thread.rs +++ b/codex-rs/thread-store/src/local/revert_thread.rs @@ -181,6 +181,7 @@ async fn create_replacement_recorder( ) .with_creator(source_meta.creator_user_id, source_meta.creator_account_id) .with_session_id(source_meta.session_id) + .with_attached_thread(source_meta.attached_thread) .with_rollout_id(rollout_id) .with_selected_capability_roots(source_meta.selected_capability_roots) .with_runtime_workspace_roots(source_meta.runtime_workspace_roots) diff --git a/codex-rs/thread-store/src/local/revert_thread_tests.rs b/codex-rs/thread-store/src/local/revert_thread_tests.rs index 974d58c93e..564c990ae2 100644 --- a/codex-rs/thread-store/src/local/revert_thread_tests.rs +++ b/codex-rs/thread-store/src/local/revert_thread_tests.rs @@ -272,6 +272,7 @@ async fn rollout_paths_for_thread( async fn create_paginated_thread(store: &LocalThreadStore, thread_id: ThreadId) { store .create_thread(CreateThreadParams { + attached_thread: false, creator_user_id: Some("creator-user".to_string()), creator_account_id: Some("creator-account".to_string()), session_id: thread_id.into(), diff --git a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs index 213735805b..c9e2fe2859 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs @@ -754,6 +754,7 @@ async fn paginated_realtime_items_materialize_separately_in_rollout_order() { let legacy_thread_id = ThreadId::new(); store .create_thread(CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: legacy_thread_id.into(), @@ -3068,6 +3069,7 @@ async fn create_paginated_subagent_thread( ) { store .create_thread(CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/thread-store/src/local/update_thread_metadata.rs b/codex-rs/thread-store/src/local/update_thread_metadata.rs index 3e2729b021..33fd5473ae 100644 --- a/codex-rs/thread-store/src/local/update_thread_metadata.rs +++ b/codex-rs/thread-store/src/local/update_thread_metadata.rs @@ -676,7 +676,9 @@ async fn metadata_for_missing_sqlite_row( patch.source.clone().unwrap_or(SessionSource::Unknown), ); builder.model_provider = patch.model_provider.clone(); - builder.history_mode = canonical_history_mode(store, thread_id, rollout_path).await?; + let (history_mode, attached_thread_parent_id) = + canonical_history_mode_and_attachment(store, thread_id, rollout_path).await?; + builder.history_mode = history_mode; builder.thread_source = patch.thread_source.clone().flatten(); builder.agent_nickname = patch.agent_nickname.clone().flatten(); builder.agent_role = patch.agent_role.clone().flatten(); @@ -684,21 +686,25 @@ async fn metadata_for_missing_sqlite_row( builder.cwd = patch.cwd.clone().map(normalize_cwd).unwrap_or_default(); builder.cli_version = patch.cli_version.clone(); let mut metadata = builder.build(store.config.default_model_provider_id.as_str()); + metadata.attached_thread_parent_id = attached_thread_parent_id; if rollout_path_archived { metadata.archived_at = Some(metadata.updated_at); } Ok(metadata) } -async fn canonical_history_mode( +async fn canonical_history_mode_and_attachment( store: &LocalThreadStore, thread_id: ThreadId, rollout_path: &Path, -) -> ThreadStoreResult { +) -> ThreadStoreResult<(ThreadHistoryMode, Option)> { if let Some((path, meta)) = live_writer::unpersisted_session_meta(store, thread_id).await && path == rollout_path { - return Ok(meta.history_mode); + return Ok(( + meta.history_mode, + meta.forked_from_id.filter(|_| meta.attached_thread), + )); } let session_meta = match read_session_meta_line(rollout_path).await { Ok(session_meta) => session_meta, @@ -715,7 +721,7 @@ async fn canonical_history_mode( { // The live writer retains the canonical mode selected before its deferred // SessionMeta reaches JSONL. - return Ok(history_mode); + return Ok((history_mode, None)); } return Err(ThreadStoreError::Internal { message: format!( @@ -732,7 +738,13 @@ async fn canonical_history_mode( ), }); } - Ok(session_meta.meta.history_mode) + Ok(( + session_meta.meta.history_mode, + session_meta + .meta + .forked_from_id + .filter(|_| session_meta.meta.attached_thread), + )) } fn requires_rollout_compatibility_update(patch: &ThreadMetadataPatch) -> bool { diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index c58a3c49e4..7a75565a68 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -485,6 +485,7 @@ mod tests { async fn create_metadata_records_originator_without_project() { let thread_id = ThreadId::new(); let sync = ThreadMetadataSync::for_create(&CreateThreadParams { + attached_thread: false, creator_user_id: None, creator_account_id: None, session_id: thread_id.into(), diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 163833a7cf..4a55334b10 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -86,6 +86,9 @@ pub struct CreateThreadParams { pub extra_config: Option, /// Source thread id when this thread is created as a fork. pub forked_from_id: Option, + /// Saved fork role, persisted independently of the thread index. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub attached_thread: bool, /// The ID of the parent thread. This will only be set if this thread is a subagent. pub parent_thread_id: Option, /// Runtime source for the thread. @@ -308,13 +311,15 @@ pub enum SortDirection { Desc, } -/// Spawn-graph relationship used to filter thread listings. +/// Parent relationship used to filter thread listings. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum ThreadRelationFilter { /// Return only threads whose immediate parent is the given thread. DirectChildrenOf(ThreadId), /// Return every thread transitively descended from the given thread. DescendantsOf(ThreadId), + /// Saved attached threads, independent of the spawned-agent graph. + AttachedThreadsOf(ThreadId), } /// Parameters for listing threads. diff --git a/codex-rs/tui/src/analytics/chats.rs b/codex-rs/tui/src/analytics/chats.rs index 50d7c8167a..ba875ecc05 100644 --- a/codex-rs/tui/src/analytics/chats.rs +++ b/codex-rs/tui/src/analytics/chats.rs @@ -145,6 +145,7 @@ pub(super) async fn roots( pub(super) fn list_params() -> ThreadListParams { ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, limit: Some(100), sort_key: Some(ThreadSortKey::UpdatedAt), sort_direction: Some(SortDirection::Desc), diff --git a/codex-rs/tui/src/app/agent_picker.rs b/codex-rs/tui/src/app/agent_picker.rs index 085fc26657..93c4b5391b 100644 --- a/codex-rs/tui/src/app/agent_picker.rs +++ b/codex-rs/tui/src/app/agent_picker.rs @@ -41,6 +41,7 @@ impl App { request_id: RequestId::String(Uuid::new_v4().to_string()), params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor, limit: Some(AGENT_PICKER_PAGE_SIZE), diff --git a/codex-rs/tui/src/app/agents_overview_discovery.rs b/codex-rs/tui/src/app/agents_overview_discovery.rs index af02a2197f..fb6aedf93b 100644 --- a/codex-rs/tui/src/app/agents_overview_discovery.rs +++ b/codex-rs/tui/src/app/agents_overview_discovery.rs @@ -36,6 +36,7 @@ pub(crate) async fn list_pinned_threads( originators: None, cursor: cursor.clone(), limit: Some(100), + attached_to_thread_id: None, sort_key: Some(ThreadSortKey::SectionPosition), sort_direction: None, model_providers: Some(Vec::new()), @@ -171,6 +172,7 @@ impl SourcePage { request_id: RequestId::String(Uuid::new_v4().to_string()), params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: self.cursor.clone(), limit: Some(PAGE_SIZE), diff --git a/codex-rs/tui/src/dynamic_tools.rs b/codex-rs/tui/src/dynamic_tools.rs index 6015350a56..4b895da67b 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_id, params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: arguments.cursor.clone(), limit: Some(limit), diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index dd93ea5169..c31c2d4da2 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -915,6 +915,7 @@ fn latest_session_lookup_params( ) -> ThreadListParams { ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(1), diff --git a/codex-rs/tui/src/named_session_lookup.rs b/codex-rs/tui/src/named_session_lookup.rs index d594fc58dd..c7a37cba48 100644 --- a/codex-rs/tui/src/named_session_lookup.rs +++ b/codex-rs/tui/src/named_session_lookup.rs @@ -65,6 +65,7 @@ pub(super) async fn lookup( let response = app_server .thread_list(ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor, limit: Some(100), diff --git a/codex-rs/tui/src/resume_picker.rs b/codex-rs/tui/src/resume_picker.rs index 847954f723..bc7631c966 100644 --- a/codex-rs/tui/src/resume_picker.rs +++ b/codex-rs/tui/src/resume_picker.rs @@ -2161,6 +2161,7 @@ fn thread_list_params( ) -> ThreadListParams { ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor, limit: Some(PAGE_SIZE as u32), diff --git a/codex-rs/tui/src/task_mentions.rs b/codex-rs/tui/src/task_mentions.rs index 364e33abcd..dd8edc5a4d 100644 --- a/codex-rs/tui/src/task_mentions.rs +++ b/codex-rs/tui/src/task_mentions.rs @@ -91,6 +91,7 @@ pub(crate) fn spawn_search( )), params: ThreadListParams { excluded_thread_ids: None, + attached_to_thread_id: None, originators: None, cursor: None, limit: Some(MAX_SEARCH_RESULTS),