diff --git a/codex-rs/rollout/src/metadata.rs b/codex-rs/rollout/src/metadata.rs index 00c6e3e637..74d1165fc8 100644 --- a/codex-rs/rollout/src/metadata.rs +++ b/codex-rs/rollout/src/metadata.rs @@ -47,6 +47,7 @@ pub(crate) fn builder_from_session_meta( session_meta.meta.source.clone(), ); builder.model_provider = session_meta.meta.model_provider.clone(); + builder.parent_thread_id = session_meta.meta.parent_thread_id; builder.agent_nickname = session_meta.meta.agent_nickname.clone(); builder.agent_role = session_meta.meta.agent_role.clone(); builder.agent_path = session_meta.meta.agent_path.clone(); diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index d908ac270d..b25ae26ad7 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -1699,7 +1699,7 @@ fn thread_item_from_state_metadata(item: codex_state::ThreadMetadata) -> ThreadI .or_else(|_| serde_json::from_value(Value::String(item.source))) .unwrap_or(SessionSource::Unknown), ), - parent_thread_id: None, + parent_thread_id: item.parent_thread_id, agent_nickname: item.agent_nickname, agent_role: item.agent_role, model_provider: Some(item.model_provider), diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index d396b27ee7..0292298f21 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -14,6 +14,7 @@ use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::TurnContextItem; use codex_protocol::protocol::UserMessageEvent; use pretty_assertions::assert_eq; @@ -803,6 +804,115 @@ async fn list_threads_state_db_only_skips_jsonl_repair_scan() -> std::io::Result Ok(()) } +#[tokio::test] +async fn list_threads_state_db_preserves_parent_thread_id() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + + let runtime = codex_state::StateRuntime::init( + home.path().to_path_buf(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + runtime + .mark_backfill_complete(/*last_watermark*/ None) + .await + .expect("backfill should be complete"); + + let uuid = Uuid::from_u128(9014); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let parent_thread_id = ThreadId::new(); + let ts = "2025-01-03T15-00-00"; + let day_dir = home.path().join("sessions/2025/01/03"); + fs::create_dir_all(&day_dir)?; + let path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl")); + let rollout_line = RolloutLine { + timestamp: ts.to_string(), + item: RolloutItem::SessionMeta(SessionMetaLine { + meta: SessionMeta { + id: thread_id, + forked_from_id: None, + parent_thread_id: Some(parent_thread_id), + timestamp: ts.to_string(), + cwd: home.path().to_path_buf(), + originator: "test_originator".to_string(), + cli_version: "test_version".to_string(), + source: SessionSource::SubAgent(SubAgentSource::Review), + thread_source: None, + agent_path: None, + agent_nickname: None, + agent_role: None, + model_provider: Some(config.model_provider_id.clone()), + base_instructions: None, + dynamic_tools: None, + memory_mode: None, + }, + git: None, + }), + }; + let mut file = File::create(&path)?; + writeln!( + file, + "{}", + serde_json::to_string(&rollout_line).expect("serialize rollout") + )?; + let user_event = RolloutLine { + timestamp: ts.to_string(), + item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "Hello from user".to_string(), + ..Default::default() + })), + }; + writeln!( + file, + "{}", + serde_json::to_string(&user_event).expect("serialize user event") + )?; + file.flush()?; + + let repaired_page = RolloutRecorder::list_threads( + Some(runtime.clone()), + &config, + /*page_size*/ 10, + /*cursor*/ None, + ThreadSortKey::CreatedAt, + SortDirection::Desc, + &[], + /*model_providers*/ None, + /*cwd_filters*/ None, + config.model_provider_id.as_str(), + /*search_term*/ None, + ) + .await?; + assert_eq!(repaired_page.items.len(), 1); + assert_eq!( + repaired_page.items[0].parent_thread_id, + Some(parent_thread_id) + ); + + let state_db_only_page = RolloutRecorder::list_threads_from_state_db( + Some(runtime.clone()), + &config, + /*page_size*/ 10, + /*cursor*/ None, + ThreadSortKey::CreatedAt, + SortDirection::Desc, + &[], + /*model_providers*/ None, + /*cwd_filters*/ None, + config.model_provider_id.as_str(), + /*search_term*/ None, + ) + .await?; + assert_eq!(state_db_only_page.items.len(), 1); + assert_eq!( + state_db_only_page.items[0].parent_thread_id, + Some(parent_thread_id) + ); + Ok(()) +} + #[tokio::test] async fn list_threads_default_filter_returns_filesystem_scan_results() -> std::io::Result<()> { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/state/migrations/0036_threads_parent_thread_id.sql b/codex-rs/state/migrations/0036_threads_parent_thread_id.sql new file mode 100644 index 0000000000..2844e2aebb --- /dev/null +++ b/codex-rs/state/migrations/0036_threads_parent_thread_id.sql @@ -0,0 +1 @@ +ALTER TABLE threads ADD COLUMN parent_thread_id TEXT; diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index f64e6f7fed..ced29a7d7f 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -51,6 +51,7 @@ fn apply_session_meta_from_item(metadata: &mut ThreadMetadata, meta_line: &Sessi metadata.id = meta_line.meta.id; metadata.source = enum_to_string(&meta_line.meta.source); metadata.thread_source = meta_line.meta.thread_source; + metadata.parent_thread_id = meta_line.meta.parent_thread_id; metadata.agent_nickname = meta_line.meta.agent_nickname.clone(); metadata.agent_role = meta_line.meta.agent_role.clone(); metadata.agent_path = meta_line.meta.agent_path.clone(); @@ -514,6 +515,7 @@ mod tests { updated_at: created_at, source: "cli".to_string(), thread_source: None, + parent_thread_id: None, agent_path: None, agent_nickname: None, agent_role: None, diff --git a/codex-rs/state/src/model/thread_metadata.rs b/codex-rs/state/src/model/thread_metadata.rs index f0b8d315e5..0456642bb3 100644 --- a/codex-rs/state/src/model/thread_metadata.rs +++ b/codex-rs/state/src/model/thread_metadata.rs @@ -71,6 +71,8 @@ pub struct ThreadMetadata { pub source: String, /// Optional analytics source classification for this thread. pub thread_source: Option, + /// Immediate control/spawn parent thread id, if this thread is a subagent. + pub parent_thread_id: Option, /// Optional random unique nickname assigned to an AgentControl-spawned sub-agent. pub agent_nickname: Option, /// Optional role (agent_role) assigned to an AgentControl-spawned sub-agent. @@ -124,6 +126,8 @@ pub struct ThreadMetadataBuilder { pub source: SessionSource, /// Optional analytics source classification for this thread. pub thread_source: Option, + /// Immediate control/spawn parent thread id, if this thread is a subagent. + pub parent_thread_id: Option, /// Optional random unique nickname assigned to the session. pub agent_nickname: Option, /// Optional role (agent_role) assigned to the session. @@ -165,6 +169,7 @@ impl ThreadMetadataBuilder { updated_at: None, source, thread_source: None, + parent_thread_id: None, agent_nickname: None, agent_role: None, agent_path: None, @@ -197,6 +202,7 @@ impl ThreadMetadataBuilder { updated_at, source, thread_source: self.thread_source, + parent_thread_id: self.parent_thread_id, agent_nickname: self.agent_nickname.clone(), agent_role: self.agent_role.clone(), agent_path: self @@ -257,6 +263,9 @@ impl ThreadMetadata { if self.source != other.source { diffs.push("source"); } + if self.parent_thread_id != other.parent_thread_id { + diffs.push("parent_thread_id"); + } if self.agent_nickname != other.agent_nickname { diffs.push("agent_nickname"); } @@ -327,6 +336,7 @@ pub(crate) struct ThreadRow { updated_at: i64, source: String, thread_source: Option, + parent_thread_id: Option, agent_nickname: Option, agent_role: Option, agent_path: Option, @@ -356,6 +366,7 @@ impl ThreadRow { updated_at: row.try_get("updated_at")?, source: row.try_get("source")?, thread_source: row.try_get("thread_source")?, + parent_thread_id: row.try_get("parent_thread_id")?, agent_nickname: row.try_get("agent_nickname")?, agent_role: row.try_get("agent_role")?, agent_path: row.try_get("agent_path")?, @@ -389,6 +400,7 @@ impl TryFrom for ThreadMetadata { updated_at, source, thread_source, + parent_thread_id, agent_nickname, agent_role, agent_path, @@ -412,6 +424,7 @@ impl TryFrom for ThreadMetadata { .map(|thread_source| thread_source.parse()) .transpose() .map_err(anyhow::Error::msg)?; + let parent_thread_id = parent_thread_id.map(ThreadId::try_from).transpose()?; Ok(Self { id: ThreadId::try_from(id)?, rollout_path: PathBuf::from(rollout_path), @@ -419,6 +432,7 @@ impl TryFrom for ThreadMetadata { updated_at: epoch_millis_to_datetime(updated_at)?, source, thread_source, + parent_thread_id, agent_nickname, agent_role, agent_path, @@ -506,6 +520,7 @@ mod tests { updated_at: 1_700_000_100, source: "cli".to_string(), thread_source: None, + parent_thread_id: None, agent_nickname: None, agent_role: None, agent_path: None, @@ -536,6 +551,7 @@ mod tests { updated_at: DateTime::::from_timestamp(1_700_000_100, 0).expect("timestamp"), source: "cli".to_string(), thread_source: None, + parent_thread_id: None, agent_nickname: None, agent_role: None, agent_path: None, @@ -568,6 +584,21 @@ mod tests { ); } + #[test] + fn thread_row_parses_parent_thread_id() { + let parent_thread_id = "00000000-0000-0000-0000-000000000124"; + let mut row = thread_row(/*reasoning_effort*/ None); + row.parent_thread_id = Some(parent_thread_id.to_string()); + let mut expected = expected_thread_metadata(/*reasoning_effort*/ None); + expected.parent_thread_id = + Some(ThreadId::from_string(parent_thread_id).expect("valid thread id")); + + assert_eq!( + ThreadMetadata::try_from(row).expect("thread metadata should parse"), + expected + ); + } + #[test] fn thread_row_ignores_unknown_reasoning_effort_values() { let metadata = ThreadMetadata::try_from(thread_row(Some("future"))) diff --git a/codex-rs/state/src/runtime/memories.rs b/codex-rs/state/src/runtime/memories.rs index 3101500796..800459315c 100644 --- a/codex-rs/state/src/runtime/memories.rs +++ b/codex-rs/state/src/runtime/memories.rs @@ -173,6 +173,7 @@ SELECT threads.updated_at_ms AS updated_at, threads.source, threads.thread_source, + threads.parent_thread_id, threads.agent_path, threads.agent_nickname, threads.agent_role, @@ -543,6 +544,7 @@ SELECT threads.updated_at_ms AS updated_at, threads.source, threads.thread_source, + threads.parent_thread_id, threads.agent_nickname, threads.agent_role, threads.agent_path, diff --git a/codex-rs/state/src/runtime/test_support.rs b/codex-rs/state/src/runtime/test_support.rs index 848a193335..b4cf9f6e19 100644 --- a/codex-rs/state/src/runtime/test_support.rs +++ b/codex-rs/state/src/runtime/test_support.rs @@ -49,6 +49,7 @@ pub(super) fn test_thread_metadata( updated_at: now, source: "cli".to_string(), thread_source: None, + parent_thread_id: None, agent_nickname: None, agent_role: None, agent_path: None, diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 9522b6176d..760c01aa9e 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -14,6 +14,7 @@ SELECT threads.updated_at_ms AS updated_at, threads.source, threads.thread_source, + threads.parent_thread_id, threads.agent_nickname, threads.agent_role, threads.agent_path, @@ -487,6 +488,7 @@ INSERT INTO threads ( updated_at_ms, source, thread_source, + parent_thread_id, agent_nickname, agent_role, agent_path, @@ -507,7 +509,7 @@ INSERT INTO threads ( git_branch, git_origin_url, memory_mode -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO NOTHING "#, ) @@ -523,6 +525,7 @@ ON CONFLICT(id) DO NOTHING .thread_source .map(codex_protocol::protocol::ThreadSource::as_str), ) + .bind(metadata.parent_thread_id.as_ref().map(ToString::to_string)) .bind(metadata.agent_nickname.as_deref()) .bind(metadata.agent_role.as_deref()) .bind(metadata.agent_path.as_deref()) @@ -693,6 +696,7 @@ INSERT INTO threads ( updated_at_ms, source, thread_source, + parent_thread_id, agent_nickname, agent_role, agent_path, @@ -713,7 +717,7 @@ INSERT INTO threads ( git_branch, git_origin_url, memory_mode -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET rollout_path = excluded.rollout_path, created_at = excluded.created_at, @@ -722,6 +726,7 @@ ON CONFLICT(id) DO UPDATE SET updated_at_ms = excluded.updated_at_ms, source = excluded.source, thread_source = excluded.thread_source, + parent_thread_id = excluded.parent_thread_id, agent_nickname = excluded.agent_nickname, agent_role = excluded.agent_role, agent_path = excluded.agent_path, @@ -755,6 +760,7 @@ ON CONFLICT(id) DO UPDATE SET .thread_source .map(codex_protocol::protocol::ThreadSource::as_str), ) + .bind(metadata.parent_thread_id.as_ref().map(ToString::to_string)) .bind(metadata.agent_nickname.as_deref()) .bind(metadata.agent_role.as_deref()) .bind(metadata.agent_path.as_deref()) @@ -925,6 +931,7 @@ SELECT threads.updated_at_ms AS updated_at, threads.source, threads.thread_source, + threads.parent_thread_id, threads.agent_nickname, threads.agent_role, threads.agent_path,