[codex-analytics] add session source to client metadata (#17374)

## Summary

Adds `thread_source` field to the existing Codex turn metadata sent to
Responses API
- Sends `thread_source: "user"` for user-initiated sessions: CLI, VS
Code, and Exec
- Sends `thread_source: "subagent"` for subagent sessions
- Omits `thread_source` for MCP, custom, and unknown session sources
- Uses the existing turn metadata transport:
  - HTTP requests send through the `x-codex-turn-metadata` header
- WebSocket `response.create` requests send through
`client_metadata["x-codex-turn-metadata"]`

## Testing
- `cargo test -p codex-protocol
session_source_thread_source_name_classifies_user_and_subagent_sources`
- `cargo test -p codex-core turn_metadata_state`
- `cargo test -p codex-core --test responses_headers
responses_stream_includes_turn_metadata_header_for_git_workspace_e2e --
--nocapture`

codex-rs/core/src/codex.rs

@@ -1570,6 +1570,7 @@ impl Session {
         let per_turn_config = Arc::new(per_turn_config);
         let turn_metadata_state = Arc::new(TurnMetadataState::new(
             conversation_id.to_string(),
+            &session_source,
             sub_id.clone(),
             cwd.to_path_buf(),
             session_configuration.sandbox_policy.get(),
@@ -5983,6 +5984,7 @@ async fn spawn_review_thread(
     let review_turn_id = sub_id.to_string();
     let turn_metadata_state = Arc::new(TurnMetadataState::new(
         sess.conversation_id.to_string(),
+        &session_source,
         review_turn_id.clone(),
         parent_turn_context.cwd.to_path_buf(),
         parent_turn_context.sandbox_policy.get(),

codex-rs/core/src/turn_metadata.rs

@@ -17,6 +17,7 @@ use codex_git_utils::get_has_changes;
 use codex_git_utils::get_head_commit_hash;
 use codex_protocol::config_types::WindowsSandboxLevel;
 use codex_protocol::protocol::SandboxPolicy;
+use codex_protocol::protocol::SessionSource;
 
 #[derive(Clone, Debug, Default)]
 struct WorkspaceGitMetadata {
@@ -58,6 +59,8 @@ pub(crate) struct TurnMetadataBag {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     session_id: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
+    thread_source: Option<&'static str>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     turn_id: Option<String>,
     #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
     workspaces: BTreeMap<String, TurnMetadataWorkspace>,
@@ -87,6 +90,7 @@ fn merge_responsesapi_client_metadata(
 
 fn build_turn_metadata_bag(
     session_id: Option<String>,
+    thread_source: Option<&'static str>,
     turn_id: Option<String>,
     sandbox: Option<String>,
     repo_root: Option<String>,
@@ -101,6 +105,7 @@ fn build_turn_metadata_bag(
 
     TurnMetadataBag {
         session_id,
+        thread_source,
         turn_id,
         workspaces,
         sandbox,
@@ -126,6 +131,7 @@ pub async fn build_turn_metadata_header(cwd: &Path, sandbox: Option<&str>) -> Op
 
     build_turn_metadata_bag(
         /*session_id*/ None,
+        /*thread_source*/ None,
         /*turn_id*/ None,
         sandbox.map(ToString::to_string),
         repo_root,
@@ -152,6 +158,7 @@ pub(crate) struct TurnMetadataState {
 impl TurnMetadataState {
     pub(crate) fn new(
         session_id: String,
+        session_source: &SessionSource,
         turn_id: String,
         cwd: PathBuf,
         sandbox_policy: &SandboxPolicy,
@@ -161,6 +168,7 @@ impl TurnMetadataState {
         let sandbox = Some(sandbox_tag(sandbox_policy, windows_sandbox_level).to_string());
         let base_metadata = build_turn_metadata_bag(
             Some(session_id),
+            session_source.thread_source_name(),
             Some(turn_id),
             sandbox,
             /*repo_root*/ None,
@@ -240,6 +248,7 @@ impl TurnMetadataState {
 
             let enriched_metadata = build_turn_metadata_bag(
                 state.base_metadata.session_id.clone(),
+                state.base_metadata.thread_source,
                 state.base_metadata.turn_id.clone(),
                 state.base_metadata.sandbox.clone(),
                 Some(repo_root),

codex-rs/core/src/turn_metadata_tests.rs

@@ -1,5 +1,7 @@
 use super::*;
 
+use codex_protocol::protocol::SessionSource;
+use codex_protocol::protocol::SubAgentSource;
 use serde_json::Value;
 use std::collections::HashMap;
 use tempfile::TempDir;
@@ -69,6 +71,7 @@ fn turn_metadata_state_uses_platform_sandbox_tag() {
 
     let state = TurnMetadataState::new(
         "session-a".to_string(),
+        &SessionSource::Exec,
         "turn-a".to_string(),
         cwd,
         &sandbox_policy,
@@ -79,10 +82,36 @@ fn turn_metadata_state_uses_platform_sandbox_tag() {
     let json: Value = serde_json::from_str(&header).expect("json");
     let sandbox_name = json.get("sandbox").and_then(Value::as_str);
     let session_id = json.get("session_id").and_then(Value::as_str);
+    let thread_source = json.get("thread_source").and_then(Value::as_str);
 
     let expected_sandbox = sandbox_tag(&sandbox_policy, WindowsSandboxLevel::Disabled);
     assert_eq!(sandbox_name, Some(expected_sandbox));
     assert_eq!(session_id, Some("session-a"));
+    assert_eq!(thread_source, Some("user"));
+    assert!(json.get("session_source").is_none());
+}
+
+#[test]
+fn turn_metadata_state_classifies_subagent_thread_source() {
+    let temp_dir = TempDir::new().expect("temp dir");
+    let cwd = temp_dir.path().to_path_buf();
+    let sandbox_policy = SandboxPolicy::new_read_only_policy();
+    let session_source = SessionSource::SubAgent(SubAgentSource::Review);
+
+    let state = TurnMetadataState::new(
+        "session-a".to_string(),
+        &session_source,
+        "turn-a".to_string(),
+        cwd,
+        &sandbox_policy,
+        WindowsSandboxLevel::Disabled,
+    );
+
+    let header = state.current_header_value().expect("header");
+    let json: Value = serde_json::from_str(&header).expect("json");
+
+    assert_eq!(json["thread_source"].as_str(), Some("subagent"));
+    assert!(json.get("session_source").is_none());
 }
 
 #[test]
@@ -93,6 +122,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields(
 
     let state = TurnMetadataState::new(
         "session-a".to_string(),
+        &SessionSource::Exec,
         "turn-a".to_string(),
         cwd,
         &sandbox_policy,
@@ -101,6 +131,7 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields(
     state.set_responsesapi_client_metadata(HashMap::from([
         ("fiber_run_id".to_string(), "fiber-123".to_string()),
         ("session_id".to_string(), "client-supplied".to_string()),
+        ("thread_source".to_string(), "client-supplied".to_string()),
     ]));
 
     let header = state.current_header_value().expect("header");
@@ -108,5 +139,6 @@ fn turn_metadata_state_merges_client_metadata_without_replacing_reserved_fields(
 
     assert_eq!(json["fiber_run_id"].as_str(), Some("fiber-123"));
     assert_eq!(json["session_id"].as_str(), Some("session-a"));
+    assert_eq!(json["thread_source"].as_str(), Some("user"));
     assert_eq!(json["turn_id"].as_str(), Some("turn-a"));
 }