diff --git a/codex-rs/app-server/tests/suite/v2/client_metadata.rs b/codex-rs/app-server/tests/suite/v2/client_metadata.rs index 80f41d7b1..3a00a433d 100644 --- a/codex-rs/app-server/tests/suite/v2/client_metadata.rs +++ b/codex-rs/app-server/tests/suite/v2/client_metadata.rs @@ -1,8 +1,16 @@ use anyhow::Result; use app_test_support::McpProcess; +use app_test_support::create_fake_rollout; use app_test_support::to_response; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ReviewDelivery; +use codex_app_server_protocol::ReviewStartParams; +use codex_app_server_protocol::ReviewStartResponse; +use codex_app_server_protocol::ReviewTarget; +use codex_app_server_protocol::ThreadForkParams; +use codex_app_server_protocol::ThreadForkResponse; +use codex_app_server_protocol::ThreadSource; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::TurnStartParams; @@ -60,6 +68,7 @@ async fn turn_start_forwards_client_metadata_to_responses_request_v2() -> Result let client_metadata = HashMap::from([ ("fiber_run_id".to_string(), "fiber-start-123".to_string()), ("origin".to_string(), "gaas".to_string()), + ("thread_source".to_string(), "client-supplied".to_string()), ]); let turn_req = mcp .send_turn_start_request(TurnStartParams { @@ -93,12 +102,188 @@ async fn turn_start_forwards_client_metadata_to_responses_request_v2() -> Result .unwrap_or_else(|| panic!("missing x-codex-turn-metadata header")); assert_eq!(metadata["fiber_run_id"].as_str(), Some("fiber-start-123")); assert_eq!(metadata["origin"].as_str(), Some("gaas")); + assert_eq!(metadata["thread_source"].as_str(), Some("client-supplied")); assert_eq!(metadata["turn_id"].as_str(), Some(turn.id.as_str())); assert!(metadata.get("session_id").is_some()); Ok(()) } +#[tokio::test] +async fn turn_start_sends_fork_lineage_in_turn_metadata_for_thread_fork_v2() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + let response_mock = responses::mount_sse_once( + &server, + responses::sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_assistant_message("msg-1", "Done"), + responses::ev_completed("resp-1"), + ]), + ) + .await; + + let codex_home = TempDir::new()?; + create_config_toml( + codex_home.path(), + &server.uri(), + /*supports_websockets*/ false, + )?; + + let source_thread_id = create_fake_rollout( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + "Saved user message", + Some("mock_provider"), + /*git_info*/ None, + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let ThreadForkResponse { thread, .. } = + fork_fake_rollout_thread(&mut mcp, source_thread_id.clone()).await?; + + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "Continue".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response::(turn_resp)?; + + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let request = response_mock.single_request(); + let metadata = request + .header("x-codex-turn-metadata") + .as_deref() + .map(parse_json_header) + .unwrap_or_else(|| panic!("missing x-codex-turn-metadata header")); + assert_eq!( + metadata["forked_from_thread_id"].as_str(), + Some(source_thread_id.as_str()) + ); + assert_eq!(metadata["thread_id"].as_str(), Some(thread.id.as_str())); + assert_eq!(metadata["turn_id"].as_str(), Some(turn.id.as_str())); + + Ok(()) +} + +#[tokio::test] +async fn review_start_sends_fork_lineage_in_turn_metadata_for_thread_fork_v2() -> Result<()> { + skip_if_no_network!(Ok(())); + + let review_payload = serde_json::json!({ + "findings": [], + "overall_correctness": "good", + "overall_explanation": "Done", + "overall_confidence_score": 0.5 + }) + .to_string(); + let server = responses::start_mock_server().await; + let response_mock = responses::mount_sse_once( + &server, + responses::sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_assistant_message("msg-1", &review_payload), + responses::ev_completed("resp-1"), + ]), + ) + .await; + + let codex_home = TempDir::new()?; + create_config_toml( + codex_home.path(), + &server.uri(), + /*supports_websockets*/ false, + )?; + + let source_thread_id = create_fake_rollout( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + "Saved user message", + Some("mock_provider"), + /*git_info*/ None, + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let ThreadForkResponse { thread, .. } = + fork_fake_rollout_thread(&mut mcp, source_thread_id.clone()).await?; + + let review_req = mcp + .send_review_start_request(ReviewStartParams { + thread_id: thread.id.clone(), + delivery: Some(ReviewDelivery::Inline), + target: ReviewTarget::Custom { + instructions: "Review the fork".to_string(), + }, + }) + .await?; + let review_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(review_req)), + ) + .await??; + let ReviewStartResponse { + review_thread_id, .. + } = to_response::(review_resp)?; + assert_eq!(review_thread_id, thread.id); + + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let request = response_mock.single_request(); + let metadata = request + .header("x-codex-turn-metadata") + .as_deref() + .map(parse_json_header) + .unwrap_or_else(|| panic!("missing x-codex-turn-metadata header")); + assert_eq!( + request.header("x-openai-subagent").as_deref(), + Some("review") + ); + assert_eq!( + metadata["forked_from_thread_id"].as_str(), + Some(review_thread_id.as_str()) + ); + let review_request_thread_id = metadata["thread_id"] + .as_str() + .unwrap_or_else(|| panic!("missing review request thread_id")); + assert!(review_request_thread_id != review_thread_id.as_str()); + assert_eq!( + request + .header("x-codex-window-id") + .as_deref() + .and_then(|window_id| window_id.split_once(':').map(|(thread_id, _)| thread_id)), + Some(review_request_thread_id) + ); + assert!(metadata["turn_id"].as_str().is_some()); + + Ok(()) +} + #[tokio::test] async fn turn_steer_updates_client_metadata_on_follow_up_responses_request_v2() -> Result<()> { skip_if_no_network!(Ok(())); @@ -345,6 +530,25 @@ supports_websockets = {supports_websockets} ) } +async fn fork_fake_rollout_thread( + mcp: &mut McpProcess, + source_thread_id: String, +) -> Result { + let fork_req = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: source_thread_id, + thread_source: Some(ThreadSource::User), + ..Default::default() + }) + .await?; + let fork_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(fork_req)), + ) + .await??; + to_response::(fork_resp) +} + fn parse_json_header(value: &str) -> serde_json::Value { match serde_json::from_str(value) { Ok(value) => value, diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index a9a90be54..4706a455f 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -262,10 +262,12 @@ impl AgentControl { .await? } (Some(session_source), None) => { + let forked_from_thread_id = thread_spawn_parent_thread_id(&session_source); Box::pin(state.spawn_new_thread_with_source( config.clone(), self.clone(), session_source, + forked_from_thread_id, /*thread_source*/ Some(ThreadSource::Subagent), /*persist_extended_history*/ false, /*metrics_service_name*/ None, @@ -487,6 +489,7 @@ impl AgentControl { self.clone(), session_source, /*thread_source*/ Some(ThreadSource::Subagent), + /*forked_from_thread_id*/ Some(parent_thread_id), /*persist_extended_history*/ false, inherited_shell_snapshot, inherited_exec_policy, diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index b27cf745c..dd88e9238 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -86,6 +86,7 @@ pub(crate) async fn run_codex_thread_interactive( extensions: Arc::clone(&parent_session.services.extensions), conversation_history: initial_history.unwrap_or(InitialHistory::New), session_source: SessionSource::SubAgent(subagent_source.clone()), + forked_from_thread_id: Some(parent_session.conversation_id), thread_source: Some(ThreadSource::Subagent), agent_control: parent_session.services.agent_control.clone(), dynamic_tools: Vec::new(), diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 1876cbf98..5ea30e5bb 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -400,6 +400,7 @@ pub(crate) struct CodexSpawnArgs { pub(crate) extensions: Arc>, pub(crate) conversation_history: InitialHistory, pub(crate) session_source: SessionSource, + pub(crate) forked_from_thread_id: Option, pub(crate) thread_source: Option, pub(crate) agent_control: AgentControl, pub(crate) dynamic_tools: Vec, @@ -464,6 +465,7 @@ impl Codex { extensions, conversation_history, session_source, + forked_from_thread_id, thread_source, agent_control, dynamic_tools, @@ -619,6 +621,7 @@ impl Codex { app_server_client_name: None, app_server_client_version: None, session_source, + forked_from_thread_id, thread_source, dynamic_tools, persist_extended_history, diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index 32122873c..762ebc145 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -69,12 +69,17 @@ pub(super) async fn spawn_review_thread( .model_reasoning_summary .unwrap_or(model_info.default_reasoning_summary); let session_source = parent_turn_context.session_source.clone(); + let forked_from_thread_id = { + let state = sess.state.lock().await; + state.session_configuration.forked_from_thread_id + }; let per_turn_config = Arc::new(per_turn_config); let review_turn_id = sub_id.to_string(); let turn_metadata_state = Arc::new(TurnMetadataState::new( sess.session_id().to_string(), sess.thread_id().to_string(), + forked_from_thread_id, parent_turn_context.thread_source, review_turn_id.clone(), #[allow(deprecated)] diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 75eb7d065..0823e9366 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -95,6 +95,8 @@ pub(crate) struct SessionConfiguration { pub(super) app_server_client_version: Option, /// Source of the session (cli, vscode, exec, mcp, ...) pub(super) session_source: SessionSource, + /// Immediate history source copied into this thread, when this thread was forked. + pub(super) forked_from_thread_id: Option, /// Optional analytics source classification for this thread. pub(super) thread_source: Option, pub(super) dynamic_tools: Vec, @@ -505,7 +507,10 @@ impl Session { session_configuration.collaboration_mode.model(), session_configuration.provider ); - let forked_from_id = initial_history.forked_from_id(); + let forked_from_id = session_configuration + .forked_from_thread_id + .or_else(|| initial_history.forked_from_id()); + session_configuration.forked_from_thread_id = forked_from_id; let event_persistence_mode = if session_configuration.persist_extended_history { ThreadEventPersistenceMode::Extended diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 063816842..5cb8f438a 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -3001,6 +3001,7 @@ async fn set_rate_limits_retains_previous_credits() { app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, + forked_from_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), persist_extended_history: false, @@ -3105,6 +3106,7 @@ async fn set_rate_limits_updates_plan_type_when_present() { app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, + forked_from_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), persist_extended_history: false, @@ -3632,6 +3634,7 @@ pub(crate) async fn make_session_configuration_for_tests() -> SessionConfigurati app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, + forked_from_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), persist_extended_history: false, @@ -4375,6 +4378,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, + forked_from_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), persist_extended_history: false, @@ -4484,6 +4488,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, + forked_from_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), persist_extended_history: false, @@ -4718,6 +4723,7 @@ async fn make_session_with_config_and_rx( app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, + forked_from_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), persist_extended_history: false, @@ -4821,6 +4827,7 @@ async fn make_session_with_history_source_and_agent_control_and_rx( app_server_client_name: None, app_server_client_version: None, session_source: session_source.clone(), + forked_from_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), persist_extended_history: false, @@ -6316,6 +6323,7 @@ where app_server_client_name: None, app_server_client_version: None, session_source: SessionSource::Exec, + forked_from_thread_id: None, thread_source: None, dynamic_tools, persist_extended_history: false, diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index 5db4aab98..dabf9c360 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -692,6 +692,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { session_source: SessionSource::SubAgent(SubAgentSource::Other( GUARDIAN_REVIEWER_NAME.to_string(), )), + forked_from_thread_id: None, thread_source: None, agent_control: AgentControl::default(), dynamic_tools: Vec::new(), diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index de3808313..df1502274 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -482,6 +482,7 @@ impl Session { let turn_metadata_state = Arc::new(TurnMetadataState::new( session_id.to_string(), thread_id.to_string(), + session_configuration.forked_from_thread_id, session_configuration.thread_source, sub_id.clone(), cwd.clone(), diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 232f12ca7..d209026b6 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -581,6 +581,15 @@ impl ThreadManager { pub async fn start_thread_with_options( &self, options: StartThreadOptions, + ) -> CodexResult { + self.start_thread_with_options_and_fork_source(options, /*forked_from_thread_id*/ None) + .await + } + + async fn start_thread_with_options_and_fork_source( + &self, + options: StartThreadOptions, + forked_from_thread_id: Option, ) -> CodexResult { let session_source = options .session_source @@ -594,6 +603,7 @@ impl ThreadManager { Arc::clone(&self.state.auth_manager), self.agent_control(), session_source, + forked_from_thread_id, thread_source, options.dynamic_tools, options.persist_extended_history, @@ -634,7 +644,8 @@ impl ThreadManager { history, InterruptedTurnHistoryMarker::from_config(&options.config), ); - self.start_thread_with_options(options).await + self.start_thread_with_options_and_fork_source(options, Some(forked_from_thread_id)) + .await } pub async fn resume_thread_from_rollout( @@ -673,6 +684,7 @@ impl ThreadManager { initial_history, auth_manager, self.agent_control(), + /*forked_from_thread_id*/ None, thread_source, Vec::new(), persist_extended_history, @@ -698,6 +710,7 @@ impl ThreadManager { InitialHistory::New, Arc::clone(&self.state.auth_manager), self.agent_control(), + /*forked_from_thread_id*/ None, /*thread_source*/ None, Vec::new(), /*persist_extended_history*/ false, @@ -727,6 +740,7 @@ impl ThreadManager { initial_history, auth_manager, self.agent_control(), + /*forked_from_thread_id*/ None, thread_source, Vec::new(), /*persist_extended_history*/ false, @@ -876,6 +890,13 @@ impl ThreadManager { persist_extended_history: bool, parent_trace: Option, ) -> CodexResult { + // `forked_from_id()` describes this history's existing lineage. When + // forking a resumed thread, the child copies the resumed thread itself. + let forked_from_thread_id = match &history { + InitialHistory::Resumed(resumed) => Some(resumed.conversation_id), + InitialHistory::Forked(_) => history.forked_from_id(), + InitialHistory::New | InitialHistory::Cleared => None, + }; let interrupted_marker = InterruptedTurnHistoryMarker::from_config(&config); let history = fork_history_from_snapshot(snapshot, history, interrupted_marker); let environments = default_thread_environment_selections( @@ -887,6 +908,7 @@ impl ThreadManager { history, Arc::clone(&self.state.auth_manager), self.agent_control(), + forked_from_thread_id, thread_source, Vec::new(), persist_extended_history, @@ -1020,6 +1042,7 @@ impl ThreadManagerState { config, agent_control, self.session_source.clone(), + /*forked_from_thread_id*/ None, /*thread_source*/ None, /*persist_extended_history*/ false, /*metrics_service_name*/ None, @@ -1036,6 +1059,7 @@ impl ThreadManagerState { config: Config, agent_control: AgentControl, session_source: SessionSource, + forked_from_thread_id: Option, thread_source: Option, persist_extended_history: bool, metrics_service_name: Option, @@ -1052,6 +1076,7 @@ impl ThreadManagerState { Arc::clone(&self.auth_manager), agent_control, session_source, + forked_from_thread_id, thread_source, Vec::new(), persist_extended_history, @@ -1086,6 +1111,7 @@ impl ThreadManagerState { Arc::clone(&self.auth_manager), agent_control, session_source, + /*forked_from_thread_id*/ None, thread_source, Vec::new(), /*persist_extended_history*/ false, @@ -1107,6 +1133,7 @@ impl ThreadManagerState { agent_control: AgentControl, session_source: SessionSource, thread_source: Option, + forked_from_thread_id: Option, persist_extended_history: bool, inherited_shell_snapshot: Option>, inherited_exec_policy: Option>, @@ -1121,6 +1148,7 @@ impl ThreadManagerState { Arc::clone(&self.auth_manager), agent_control, session_source, + forked_from_thread_id, thread_source, Vec::new(), persist_extended_history, @@ -1142,6 +1170,7 @@ impl ThreadManagerState { initial_history: InitialHistory, auth_manager: Arc, agent_control: AgentControl, + forked_from_thread_id: Option, thread_source: Option, dynamic_tools: Vec, persist_extended_history: bool, @@ -1156,6 +1185,7 @@ impl ThreadManagerState { auth_manager, agent_control, self.session_source.clone(), + forked_from_thread_id, thread_source, dynamic_tools, persist_extended_history, @@ -1177,6 +1207,7 @@ impl ThreadManagerState { auth_manager: Arc, agent_control: AgentControl, session_source: SessionSource, + forked_from_thread_id: Option, thread_source: Option, dynamic_tools: Vec, persist_extended_history: bool, @@ -1229,6 +1260,7 @@ impl ThreadManagerState { extensions: Arc::clone(&self.extensions), conversation_history: initial_history, session_source, + forked_from_thread_id, thread_source, agent_control, dynamic_tools, diff --git a/codex-rs/core/src/turn_metadata.rs b/codex-rs/core/src/turn_metadata.rs index 1b2c6b4b1..52a714960 100644 --- a/codex-rs/core/src/turn_metadata.rs +++ b/codex-rs/core/src/turn_metadata.rs @@ -16,6 +16,7 @@ use codex_git_utils::get_git_remote_urls_assume_git_repo; use codex_git_utils::get_git_repo_root; use codex_git_utils::get_has_changes; use codex_git_utils::get_head_commit_hash; +use codex_protocol::ThreadId; use codex_protocol::config_types::WindowsSandboxLevel; use codex_protocol::models::PermissionProfile; use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig; @@ -74,6 +75,8 @@ pub(crate) struct TurnMetadataBag { #[serde(default, skip_serializing_if = "Option::is_none")] thread_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] + forked_from_thread_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] thread_source: Option, #[serde(default, skip_serializing_if = "Option::is_none")] turn_id: Option, @@ -107,7 +110,14 @@ fn merge_turn_metadata( } if let Some(responsesapi_client_metadata) = responsesapi_client_metadata { for (key, value) in responsesapi_client_metadata { - if key == TURN_STARTED_AT_UNIX_MS_KEY { + if matches!( + key.as_str(), + "session_id" + | "thread_id" + | "turn_id" + | TURN_STARTED_AT_UNIX_MS_KEY + | "forked_from_thread_id" + ) { continue; } metadata @@ -118,32 +128,6 @@ fn merge_turn_metadata( to_ascii_json_string(&metadata).ok() } -fn build_turn_metadata_bag( - session_id: Option, - thread_id: Option, - thread_source: Option, - turn_id: Option, - sandbox: Option, - repo_root: Option, - workspace_git_metadata: Option, -) -> TurnMetadataBag { - let mut workspaces = BTreeMap::new(); - if let (Some(repo_root), Some(workspace_git_metadata)) = (repo_root, workspace_git_metadata) - && !workspace_git_metadata.is_empty() - { - workspaces.insert(repo_root, workspace_git_metadata.into()); - } - - TurnMetadataBag { - session_id, - thread_id, - thread_source, - turn_id, - workspaces, - sandbox, - } -} - pub async fn build_turn_metadata_header( cwd: &AbsolutePathBuf, sandbox: Option<&str>, @@ -164,20 +148,24 @@ pub async fn build_turn_metadata_header( return None; } - build_turn_metadata_bag( - /*session_id*/ None, - /*thread_id*/ None, - /*thread_source*/ None, - /*turn_id*/ None, - sandbox.map(ToString::to_string), - repo_root, - Some(WorkspaceGitMetadata { - associated_remote_urls, - latest_git_commit_hash, - has_changes, - }), - ) - .to_header_value() + let workspace_git_metadata = WorkspaceGitMetadata { + associated_remote_urls, + latest_git_commit_hash, + has_changes, + }; + let mut metadata = TurnMetadataBag { + sandbox: sandbox.map(ToString::to_string), + ..Default::default() + }; + if let Some(repo_root) = repo_root + && !workspace_git_metadata.is_empty() + { + metadata + .workspaces + .insert(repo_root, workspace_git_metadata.into()); + } + + metadata.to_header_value() } #[derive(Clone, Debug)] @@ -198,6 +186,7 @@ impl TurnMetadataState { pub(crate) fn new( session_id: String, thread_id: String, + forked_from_thread_id: Option, thread_source: Option, turn_id: String, cwd: AbsolutePathBuf, @@ -214,15 +203,15 @@ impl TurnMetadataState { ) .to_string(), ); - let base_metadata = build_turn_metadata_bag( - Some(session_id), - Some(thread_id), + let base_metadata = TurnMetadataBag { + session_id: Some(session_id), + thread_id: Some(thread_id), + forked_from_thread_id, thread_source, - Some(turn_id), + turn_id: Some(turn_id), sandbox, - /*repo_root*/ None, - /*workspace_git_metadata*/ None, - ); + ..Default::default() + }; let base_header = base_metadata .to_header_value() .unwrap_or_else(|| "{}".to_string()); @@ -346,19 +335,13 @@ impl TurnMetadataState { let Some(repo_root) = state.repo_root.clone() else { return; }; - - let enriched_metadata = build_turn_metadata_bag( - state.base_metadata.session_id.clone(), - state.base_metadata.thread_id.clone(), - state.base_metadata.thread_source, - state.base_metadata.turn_id.clone(), - state.base_metadata.sandbox.clone(), - Some(repo_root), - Some(workspace_git_metadata), - ); - if enriched_metadata.workspaces.is_empty() { + if workspace_git_metadata.is_empty() { return; } + let mut enriched_metadata = state.base_metadata.clone(); + enriched_metadata + .workspaces + .insert(repo_root, workspace_git_metadata.into()); if let Some(header_value) = enriched_metadata.to_header_value() { *state diff --git a/codex-rs/core/src/turn_metadata_tests.rs b/codex-rs/core/src/turn_metadata_tests.rs index 2f7992710..a874b0db9 100644 --- a/codex-rs/core/src/turn_metadata_tests.rs +++ b/codex-rs/core/src/turn_metadata_tests.rs @@ -93,6 +93,7 @@ fn turn_metadata_state_uses_platform_sandbox_tag() { let state = TurnMetadataState::new( "session-a".to_string(), "thread-a".to_string(), + /*forked_from_thread_id*/ None, Some(ThreadSource::User), "turn-a".to_string(), cwd, @@ -128,6 +129,7 @@ fn turn_metadata_state_uses_explicit_subagent_thread_source() { let state = TurnMetadataState::new( "session-a".to_string(), "thread-a".to_string(), + /*forked_from_thread_id*/ None, Some(ThreadSource::Subagent), "turn-a".to_string(), cwd, @@ -143,6 +145,35 @@ fn turn_metadata_state_uses_explicit_subagent_thread_source() { assert!(json.get("session_source").is_none()); } +#[test] +fn turn_metadata_state_includes_root_fork_lineage() { + let temp_dir = TempDir::new().expect("temp dir"); + let cwd = temp_dir.path().abs(); + let permission_profile = PermissionProfile::read_only(); + let source_thread_id = + ThreadId::from_string("11111111-1111-4111-8111-111111111111").expect("thread id"); + + let state = TurnMetadataState::new( + "session-a".to_string(), + "thread-a".to_string(), + Some(source_thread_id), + Some(ThreadSource::User), + "turn-a".to_string(), + cwd, + &permission_profile, + WindowsSandboxLevel::Disabled, + /*enforce_managed_network*/ false, + ); + + let header = state.current_header_value().expect("header"); + let json: Value = serde_json::from_str(&header).expect("json"); + + assert_eq!( + json["forked_from_thread_id"].as_str(), + Some("11111111-1111-4111-8111-111111111111") + ); +} + #[test] fn turn_metadata_state_includes_turn_started_at_unix_ms_after_start() { let temp_dir = TempDir::new().expect("temp dir"); @@ -152,6 +183,7 @@ fn turn_metadata_state_includes_turn_started_at_unix_ms_after_start() { let state = TurnMetadataState::new( "session-a".to_string(), "thread-a".to_string(), + /*forked_from_thread_id*/ None, Some(ThreadSource::User), "turn-a".to_string(), cwd, @@ -179,6 +211,7 @@ fn turn_metadata_state_includes_model_and_reasoning_effort_only_in_request_meta( let state = TurnMetadataState::new( "session-a".to_string(), "thread-a".to_string(), + /*forked_from_thread_id*/ None, /*thread_source*/ None, "turn-a".to_string(), cwd, @@ -224,6 +257,7 @@ fn turn_metadata_state_marks_user_input_requested_during_turn_only_for_mcp_reque let state = TurnMetadataState::new( "session-a".to_string(), "thread-a".to_string(), + /*forked_from_thread_id*/ None, /*thread_source*/ None, "turn-a".to_string(), cwd, @@ -266,7 +300,7 @@ fn turn_metadata_state_marks_user_input_requested_during_turn_only_for_mcp_reque } #[test] -fn turn_metadata_state_ignores_client_turn_started_at_unix_ms_before_start() { +fn turn_metadata_state_ignores_client_reserved_metadata_before_start() { let temp_dir = TempDir::new().expect("temp dir"); let cwd = temp_dir.path().abs(); let permission_profile = PermissionProfile::read_only(); @@ -274,6 +308,7 @@ fn turn_metadata_state_ignores_client_turn_started_at_unix_ms_before_start() { let state = TurnMetadataState::new( "session-a".to_string(), "thread-a".to_string(), + /*forked_from_thread_id*/ None, Some(ThreadSource::User), "turn-a".to_string(), cwd, @@ -281,15 +316,22 @@ fn turn_metadata_state_ignores_client_turn_started_at_unix_ms_before_start() { WindowsSandboxLevel::Disabled, /*enforce_managed_network*/ false, ); - state.set_responsesapi_client_metadata(HashMap::from([( - "turn_started_at_unix_ms".to_string(), - "client-supplied".to_string(), - )])); + state.set_responsesapi_client_metadata(HashMap::from([ + ( + "turn_started_at_unix_ms".to_string(), + "client-supplied".to_string(), + ), + ( + "forked_from_thread_id".to_string(), + "client-supplied".to_string(), + ), + ])); let header = state.current_header_value().expect("header"); let json: Value = serde_json::from_str(&header).expect("json"); assert!(json.get("turn_started_at_unix_ms").is_none()); + assert!(json.get("forked_from_thread_id").is_none()); } #[test] @@ -297,10 +339,13 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( let temp_dir = TempDir::new().expect("temp dir"); let cwd = temp_dir.path().abs(); let permission_profile = PermissionProfile::read_only(); + let source_thread_id = + ThreadId::from_string("44444444-4444-4444-8444-444444444444").expect("thread id"); let state = TurnMetadataState::new( "session-a".to_string(), "thread-a".to_string(), + Some(source_thread_id), Some(ThreadSource::User), "turn-a".to_string(), cwd, @@ -318,6 +363,10 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( ), ("session_id".to_string(), "client-supplied".to_string()), ("thread_id".to_string(), "client-supplied".to_string()), + ( + "forked_from_thread_id".to_string(), + "client-supplied".to_string(), + ), ("thread_source".to_string(), "client-supplied".to_string()), ( "turn_started_at_unix_ms".to_string(), @@ -337,6 +386,10 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields( assert_eq!(json["reasoning_effort"].as_str(), Some("client-supplied")); assert_eq!(json["session_id"].as_str(), Some("session-a")); assert_eq!(json["thread_id"].as_str(), Some("thread-a")); + assert_eq!( + json["forked_from_thread_id"].as_str(), + Some("44444444-4444-4444-8444-444444444444") + ); assert_eq!(json["thread_source"].as_str(), Some("user")); assert_eq!(json["turn_id"].as_str(), Some("turn-a")); assert_eq!( diff --git a/codex-rs/core/tests/suite/responses_api_proxy_headers.rs b/codex-rs/core/tests/suite/responses_api_proxy_headers.rs index 809a8bb70..0c401d05f 100644 --- a/codex-rs/core/tests/suite/responses_api_proxy_headers.rs +++ b/codex-rs/core/tests/suite/responses_api_proxy_headers.rs @@ -125,6 +125,15 @@ async fn responses_api_parent_and_subagent_requests_include_identity_headers() - child.header("x-codex-parent-thread-id").as_deref(), Some(parent_thread_id) ); + let child_turn_metadata: serde_json::Value = serde_json::from_str( + &child + .header("x-codex-turn-metadata") + .ok_or_else(|| anyhow!("child request missing x-codex-turn-metadata"))?, + )?; + assert_eq!( + child_turn_metadata["forked_from_thread_id"].as_str(), + Some(parent_thread_id) + ); Ok(()) } diff --git a/codex-rs/core/tests/suite/review.rs b/codex-rs/core/tests/suite/review.rs index 802035d7d..9efaf6c2d 100644 --- a/codex-rs/core/tests/suite/review.rs +++ b/codex-rs/core/tests/suite/review.rs @@ -61,7 +61,7 @@ async fn review_op_emits_lifecycle_and_review_output() { "overall_confidence_score": 0.8 }) .to_string(); - let (server, _request_log) = start_responses_server_with_sse( + let (server, request_log) = start_responses_server_with_sse( assistant_message_sse(&review_json), /*expected_requests*/ 1, ) @@ -111,11 +111,38 @@ async fn review_op_emits_lifecycle_and_review_output() { assert_eq!(expected, review); let _complete = wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; - // Also verify that a user message with the header and a formatted finding - // was recorded back in the parent session's rollout. let path = codex.rollout_path().expect("rollout path"); let text = std::fs::read_to_string(&path).expect("read rollout file"); + let parent_thread_id = text + .lines() + .filter(|line| !line.trim().is_empty()) + .find_map(|line| { + let rollout_line: RolloutLine = serde_json::from_str(line).expect("rollout line"); + match rollout_line.item { + RolloutItem::SessionMeta(session_meta) => Some(session_meta.meta.id.to_string()), + _ => None, + } + }) + .expect("parent session meta"); + let request = request_log.single_request(); + assert_eq!( + request.header("x-openai-subagent").as_deref(), + Some("review") + ); + let turn_metadata: serde_json::Value = serde_json::from_str( + &request + .header("x-codex-turn-metadata") + .expect("review request turn metadata"), + ) + .expect("review request turn metadata json"); + assert_eq!( + turn_metadata["forked_from_thread_id"].as_str(), + Some(parent_thread_id.as_str()) + ); + + // Also verify that a user message with the header and a formatted finding + // was recorded back in the parent session's rollout. let mut saw_header = false; let mut saw_finding_line = false; let expected_assistant_text = render_review_output_text(&expected);