agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add forked_from_thread_id turn metadata (#24160)

Browse Source 
## Why

When Codex calls responsesapi, we currently send `session_id`,
`thread_id`, and `turn_id` among other things as
`client_metadata["x-codex-turn-metadata"]`. This PR adds
`forked_from_thread_id` which helps explain the "lineage" of a forked
thread.

## What's changed

- Track the immediate history source copied into a forked thread through
thread/session creation, including subagent and review turn metadata
paths.
- Include `forked_from_thread_id` in Codex turn metadata while
preventing turn-scoped Responses API client metadata from overwriting
Codex-owned lineage fields.
- Add coverage for fork lineage in turn metadata and the app-server
Responses API request path.
This commit is contained in:
Owen Lin
2026-05-26 14:05:28 -07:00
committed by GitHub
Unverified
parent 5cd9b8086a
commit 1911021c0e
14 changed files with 404 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/client_metadata.rs
+204
View File
@@ -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::<TurnStartResponse>(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::<ReviewStartResponse>(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<ThreadForkResponse> {
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::<ThreadForkResponse>(fork_resp)
}
fn parse_json_header(value: &str) -> serde_json::Value {
match serde_json::from_str(value) {
Ok(value) => value,
codex-rs/core/src/agent/control.rs
+3
View File
@@ -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,
codex-rs/core/src/codex_delegate.rs
+1
View File
@@ -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(),
codex-rs/core/src/session/mod.rs
+3
View File
@@ -400,6 +400,7 @@ pub(crate) struct CodexSpawnArgs {
pub(crate) extensions: Arc<codex_extension_api::ExtensionRegistry<crate::config::Config>>,
pub(crate) conversation_history: InitialHistory,
pub(crate) session_source: SessionSource,
pub(crate) forked_from_thread_id: Option<ThreadId>,
pub(crate) thread_source: Option<ThreadSource>,
pub(crate) agent_control: AgentControl,
pub(crate) dynamic_tools: Vec<DynamicToolSpec>,
@@ -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,
codex-rs/core/src/session/review.rs
+5
View File
@@ -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)]
codex-rs/core/src/session/session.rs
+6 -1
View File
@@ -95,6 +95,8 @@ pub(crate) struct SessionConfiguration {
pub(super) app_server_client_version: Option<String>,
/// 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<ThreadId>,
/// Optional analytics source classification for this thread.
pub(super) thread_source: Option<ThreadSource>,
pub(super) dynamic_tools: Vec<DynamicToolSpec>,
@@ -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
codex-rs/core/src/session/tests.rs
+8
View File
@@ -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,
codex-rs/core/src/session/tests/guardian_tests.rs
+1
View File
@@ -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(),
codex-rs/core/src/session/turn_context.rs
+1
View File
@@ -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(),
codex-rs/core/src/thread_manager.rs
+33 -1
View File
@@ -581,6 +581,15 @@ impl ThreadManager {
pub async fn start_thread_with_options(
&self,
options: StartThreadOptions,
) -> CodexResult<NewThread> {
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<ThreadId>,
) -> CodexResult<NewThread> {
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<W3cTraceContext>,
) -> CodexResult<NewThread> {
// `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<ThreadId>,
thread_source: Option<ThreadSource>,
persist_extended_history: bool,
metrics_service_name: Option<String>,
@@ -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<ThreadSource>,
forked_from_thread_id: Option<ThreadId>,
persist_extended_history: bool,
inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
@@ -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<AuthManager>,
agent_control: AgentControl,
forked_from_thread_id: Option<ThreadId>,
thread_source: Option<ThreadSource>,
dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
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<AuthManager>,
agent_control: AgentControl,
session_source: SessionSource,
forked_from_thread_id: Option<ThreadId>,
thread_source: Option<ThreadSource>,
dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
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,
codex-rs/core/src/turn_metadata.rs
+42 -59
View File
@@ -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<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
forked_from_thread_id: Option<ThreadId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
thread_source: Option<ThreadSource>,
#[serde(default, skip_serializing_if = "Option::is_none")]
turn_id: Option<String>,
@@ -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<String>,
thread_id: Option<String>,
thread_source: Option<ThreadSource>,
turn_id: Option<String>,
sandbox: Option<String>,
repo_root: Option<String>,
workspace_git_metadata: Option<WorkspaceGitMetadata>,
) -> 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<ThreadId>,
thread_source: Option<ThreadSource>,
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
codex-rs/core/src/turn_metadata_tests.rs
+58 -5
View File
@@ -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!(
codex-rs/core/tests/suite/responses_api_proxy_headers.rs
+9
View File
@@ -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(())
}
codex-rs/core/tests/suite/review.rs
+30 -3
View File
@@ -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);