Distinguish Guardian review threads from subagents (#40221)

## Why

Guardian reviews were reported with the generic `subagent` thread source, making them indistinguishable from other delegated threads in persisted metadata and analytics.

## What changed

- Add the `guardian_review` thread source to the core protocol and app-server schema.
- Assign it to Guardian reviewer sessions and propagate the configured source through turn metadata, rollouts, and analytics.
- Continue treating Guardian reviews like subagents for paginated history and trusted-provenance checks.

## Testing

- Cover protocol round trips, delegate source selection, emitted analytics, request metadata, and persisted rollout metadata.

GitOrigin-RevId: 07480c122715812874c9d3b48bb39fc5c86b2367

codex-rs/analytics/src/analytics_client_tests.rs

@@ -3232,6 +3232,7 @@ fn subagent_thread_started_review_serializes_expected_shape() {
             client_version: "1.0.0".to_string(),
             model: "gpt-5".to_string(),
             ephemeral: false,
+            thread_source: Some(ThreadSource::Subagent),
             subagent_source: SubAgentSource::Review,
             created_at: 123,
         },
@@ -3284,6 +3285,7 @@ fn subagent_thread_started_thread_spawn_serializes_thread_lineage() {
             client_version: "1.0.0".to_string(),
             model: "gpt-5".to_string(),
             ephemeral: true,
+            thread_source: Some(ThreadSource::Subagent),
             subagent_source: SubAgentSource::ThreadSpawn {
                 parent_thread_id,
                 depth: 1,
@@ -3323,6 +3325,7 @@ fn subagent_thread_started_memory_consolidation_serializes_expected_shape() {
             client_version: "1.0.0".to_string(),
             model: "gpt-5".to_string(),
             ephemeral: false,
+            thread_source: Some(ThreadSource::Subagent),
             subagent_source: SubAgentSource::MemoryConsolidation,
             created_at: 125,
         },
@@ -3350,12 +3353,14 @@ fn subagent_thread_started_other_serializes_expected_shape() {
             client_version: "1.0.0".to_string(),
             model: "gpt-5".to_string(),
             ephemeral: false,
+            thread_source: Some(ThreadSource::GuardianReview),
             subagent_source: SubAgentSource::Other("guardian".to_string()),
             created_at: 126,
         },
     ));
 
     let payload = serde_json::to_value(&event).expect("serialize other subagent event");
+    assert_eq!(payload["event_params"]["thread_source"], "guardian_review");
     assert_eq!(payload["event_params"]["subagent_source"], "guardian");
     assert_eq!(payload["event_params"]["parent_thread_id"], json!(null));
 }
@@ -3376,6 +3381,7 @@ fn subagent_thread_started_other_serializes_explicit_parent_thread_id() {
             client_version: "1.0.0".to_string(),
             model: "gpt-5".to_string(),
             ephemeral: false,
+            thread_source: Some(ThreadSource::GuardianReview),
             subagent_source: SubAgentSource::Other("guardian".to_string()),
             created_at: 126,
         },
@@ -3407,6 +3413,7 @@ async fn subagent_thread_started_publishes_without_initialize() {
                     client_version: "1.0.0".to_string(),
                     model: "gpt-5".to_string(),
                     ephemeral: false,
+                    thread_source: Some(ThreadSource::Subagent),
                     subagent_source: SubAgentSource::Review,
                     created_at: 127,
                 },
@@ -3427,7 +3434,7 @@ async fn subagent_thread_started_publishes_without_initialize() {
 }
 
 #[tokio::test]
-async fn subagent_events_keep_thread_originator_with_explicit_turn_connection() {
+async fn guardian_events_keep_thread_source_and_originator_with_explicit_turn_connection() {
     let mut reducer = AnalyticsReducer::default();
     let mut events = Vec::new();
     let parent_thread_id =
@@ -3483,13 +3490,8 @@ async fn subagent_events_keep_thread_originator_with_explicit_turn_connection()
                     client_version: "1.0.0".to_string(),
                     model: "gpt-5".to_string(),
                     ephemeral: false,
-                    subagent_source: SubAgentSource::ThreadSpawn {
-                        parent_thread_id,
-                        depth: 1,
-                        agent_path: None,
-                        agent_nickname: None,
-                        agent_role: None,
-                    },
+                    thread_source: Some(ThreadSource::GuardianReview),
+                    subagent_source: SubAgentSource::Other("guardian".to_string()),
                     created_at: 130,
                 },
             )),
@@ -3546,8 +3548,8 @@ async fn subagent_events_keep_thread_originator_with_explicit_turn_connection()
     };
     let params = &event.event_params;
     assert_eq!(params.session_id, "session-root");
-    assert_eq!(params.thread_source, Some(ThreadSource::Subagent));
-    assert_eq!(params.subagent_source.as_deref(), Some("thread_spawn"));
+    assert_eq!(params.thread_source, Some(ThreadSource::GuardianReview));
+    assert_eq!(params.subagent_source.as_deref(), Some("guardian"));
     assert_eq!(
         params.parent_thread_id.as_deref(),
         Some("44444444-4444-4444-4444-444444444444")
@@ -3625,6 +3627,7 @@ async fn subagent_tool_items_inherit_parent_connection_metadata() {
                     client_version: "1.0.0".to_string(),
                     model: "gpt-5".to_string(),
                     ephemeral: false,
+                    thread_source: Some(ThreadSource::Subagent),
                     subagent_source: SubAgentSource::Review,
                     created_at: 128,
                 },

codex-rs/analytics/src/events.rs

@@ -1449,7 +1449,7 @@ pub(crate) fn subagent_thread_started_event_request(
         runtime: current_runtime_metadata(),
         model: input.model,
         ephemeral: input.ephemeral,
-        thread_source: Some(ThreadSource::Subagent),
+        thread_source: input.thread_source,
         initialization_mode: ThreadInitializationMode::New,
         subagent_source: Some(subagent_source_name(&input.subagent_source)),
         parent_thread_id: input.parent_thread_id,

codex-rs/analytics/src/facts.rs

@@ -28,6 +28,7 @@ use codex_protocol::protocol::HookSource;
 use codex_protocol::protocol::SessionSource;
 use codex_protocol::protocol::SkillScope;
 use codex_protocol::protocol::SubAgentSource;
+use codex_protocol::protocol::ThreadSource;
 use codex_protocol::protocol::TokenUsage;
 use codex_protocol::request_permissions::RequestPermissionsResponse;
 use serde::Serialize;
@@ -388,6 +389,7 @@ pub struct SubAgentThreadStartedInput {
     pub client_version: String,
     pub model: String,
     pub ephemeral: bool,
+    pub thread_source: Option<ThreadSource>,
     pub subagent_source: SubAgentSource,
     pub created_at: u64,
 }

codex-rs/analytics/src/reducer.rs

@@ -1069,7 +1069,7 @@ impl AnalyticsReducer {
             .metadata
             .get_or_insert_with(|| ThreadMetadataState {
                 session_id: input.session_id.clone(),
-                thread_source: Some(ThreadSource::Subagent),
+                thread_source: input.thread_source.clone(),
                 initialization_mode: ThreadInitializationMode::New,
                 subagent_source: Some(subagent_source_name(&input.subagent_source)),
                 parent_thread_id,

codex-rs/app-server-protocol/src/protocol/v2/tests.rs

@@ -144,6 +144,7 @@ fn thread_sources_round_trip_as_scalar_labels() {
     for (source, label) in [
         (ThreadSource::User, "user"),
         (ThreadSource::Subagent, "subagent"),
+        (ThreadSource::GuardianReview, "guardian_review"),
         (
             ThreadSource::Feature("automation".to_string()),
             "automation",

codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs

@@ -101,6 +101,7 @@ impl From<ThreadHistoryMode> for CoreThreadHistoryMode {
 pub enum ThreadSource {
     User,
     Subagent,
+    GuardianReview,
     Feature(String),
     MemoryConsolidation,
 }
@@ -135,6 +136,7 @@ impl From<CoreThreadSource> for ThreadSource {
         match value {
             CoreThreadSource::User => ThreadSource::User,
             CoreThreadSource::Subagent => ThreadSource::Subagent,
+            CoreThreadSource::GuardianReview => ThreadSource::GuardianReview,
             CoreThreadSource::Feature(feature) => ThreadSource::Feature(feature),
             CoreThreadSource::MemoryConsolidation => ThreadSource::MemoryConsolidation,
         }
@@ -146,6 +148,7 @@ impl From<ThreadSource> for CoreThreadSource {
         match value {
             ThreadSource::User => CoreThreadSource::User,
             ThreadSource::Subagent => CoreThreadSource::Subagent,
+            ThreadSource::GuardianReview => CoreThreadSource::GuardianReview,
             ThreadSource::Feature(feature) => CoreThreadSource::Feature(feature),
             ThreadSource::MemoryConsolidation => CoreThreadSource::MemoryConsolidation,
         }

codex-rs/core/src/codex_delegate.rs

@@ -80,7 +80,8 @@ pub(crate) async fn run_codex_thread_interactive(
         warnings: Vec::new(),
     };
     let session_source = SessionSource::SubAgent(subagent_source.clone());
-    let extensions = if crate::guardian::is_guardian_reviewer_source(&session_source) {
+    let is_guardian_reviewer = crate::guardian::is_guardian_reviewer_source(&session_source);
+    let extensions = if is_guardian_reviewer {
         codex_extension_api::empty_extension_registry()
     } else {
         Arc::clone(&parent_session.services.extensions)
@@ -107,7 +108,11 @@ pub(crate) async fn run_codex_thread_interactive(
         session_source,
         forked_from_thread_id,
         parent_thread_id: Some(parent_session.thread_id),
-        thread_source: Some(ThreadSource::Subagent),
+        thread_source: Some(if is_guardian_reviewer {
+            ThreadSource::GuardianReview
+        } else {
+            ThreadSource::Subagent
+        }),
         originator: parent_ctx.originator.clone(),
         agent_control: parent_session.services.agent_control.clone(),
         dynamic_tools: Vec::new(),

codex-rs/core/src/codex_delegate_tests.rs

@@ -229,12 +229,13 @@ async fn guardian_delegates_do_not_inherit_parent_extensions() {
         .services
         .extensions = Arc::new(extensions.build());
 
-    for (subagent_source, expected_thread_starts) in [
+    for (subagent_source, expected_thread_starts, expected_thread_source) in [
         (
             SubAgentSource::Other(crate::guardian::GUARDIAN_REVIEWER_NAME.to_string()),
             0,
+            ThreadSource::GuardianReview,
         ),
-        (SubAgentSource::Review, 1),
+        (SubAgentSource::Review, 1, ThreadSource::Subagent),
     ] {
         let mut config = parent_ctx.config.as_ref().clone();
         config.permissions.approval_policy = Constrained::allow_only(AskForApproval::Never);
@@ -263,6 +264,10 @@ async fn guardian_delegates_do_not_inherit_parent_extensions() {
             expected_thread_starts
         );
         assert_eq!(thread_starts.load(Ordering::SeqCst), expected_thread_starts);
+        assert_eq!(
+            session.thread_config_snapshot().await.thread_source,
+            Some(expected_thread_source)
+        );
         io.shutdown_and_wait()
             .await
             .expect("delegate session should shut down");

codex-rs/core/src/session/mod.rs

@@ -1323,7 +1323,7 @@ impl Session {
                     ThreadHistoryMode::Paginated
                 ) && matches!(
                     session_configuration.thread_source.as_ref(),
-                    Some(ThreadSource::Subagent)
+                    Some(ThreadSource::Subagent | ThreadSource::GuardianReview)
                 ),
             )
         };
@@ -4230,6 +4230,7 @@ pub(crate) fn emit_subagent_session_started(
         client_version,
         model: thread_config.model,
         ephemeral: thread_config.ephemeral,
+        thread_source: thread_config.thread_source,
         subagent_source,
         created_at,
     });

codex-rs/core/src/session/session.rs

@@ -732,7 +732,7 @@ impl Session {
             ThreadHistoryMode::Paginated
         ) && matches!(
             session_configuration.thread_source.as_ref(),
-            Some(ThreadSource::Subagent)
+            Some(ThreadSource::Subagent | ThreadSource::GuardianReview)
         );
         if let InitialHistory::Forked(items) = &mut initial_history {
             Self::assign_missing_rollout_response_item_ids(items);

codex-rs/core/src/session/tests.rs

@@ -4954,6 +4954,7 @@ async fn emit_subagent_session_started_includes_fork_lineage_and_originator() {
     let child_thread_id = ThreadId::new();
     let mut session_configuration = make_session_configuration_for_tests().await;
     session_configuration.forked_from_thread_id = Some(forked_from_thread_id);
+    session_configuration.thread_source = Some(ThreadSource::GuardianReview);
 
     emit_subagent_session_started(
         &analytics_events_client,
@@ -4965,13 +4966,7 @@ async fn emit_subagent_session_started_includes_fork_lineage_and_originator() {
         child_thread_id,
         Some(parent_thread_id),
         session_configuration.thread_config_snapshot(Vec::new()),
-        SubAgentSource::ThreadSpawn {
-            parent_thread_id,
-            depth: 1,
-            agent_path: None,
-            agent_nickname: None,
-            agent_role: None,
-        },
+        SubAgentSource::Other(crate::guardian::GUARDIAN_REVIEWER_NAME.to_string()),
     );
 
     let event = timeout(Duration::from_secs(1), async {
@@ -4995,6 +4990,7 @@ async fn emit_subagent_session_started_includes_fork_lineage_and_originator() {
     .await
     .expect("subagent initialization analytics should be emitted");
 
+    assert_eq!(event["event_params"]["thread_source"], "guardian_review");
     assert_eq!(
         event["event_params"]["parent_thread_id"],
         parent_thread_id.to_string()

codex-rs/core/src/turn_metadata.rs

@@ -292,7 +292,11 @@ impl TurnMetadataState {
         }
         match &self.thread_source {
             // Desktop create/fork/send lacks trusted app-server provenance; fail closed.
-            Some(ThreadSource::Subagent | ThreadSource::MemoryConsolidation) => false,
+            Some(
+                ThreadSource::Subagent
+                | ThreadSource::GuardianReview
+                | ThreadSource::MemoryConsolidation,
+            ) => false,
             Some(ThreadSource::Feature(feature)) => {
                 !matches!(feature.as_str(), "system" | "title") && !feature.starts_with("ambient")
             }

codex-rs/core/tests/suite/guardian_review.rs

@@ -42,6 +42,7 @@ use codex_protocol::protocol::Op;
 use codex_protocol::protocol::ReviewDecision;
 use codex_protocol::protocol::SandboxPolicy;
 use codex_protocol::protocol::ThreadSettingsOverrides;
+use codex_protocol::protocol::ThreadSource;
 use codex_protocol::protocol::TurnAbortReason;
 use codex_protocol::user_input::UserInput;
 use core_test_support::fs_wait;
@@ -398,6 +399,14 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review(
         Some("guardian")
     );
     assert_eq!(guardian_review["model"].as_str(), Some(expected_model));
+    for request in [guardian_prewarm, &guardian_review] {
+        let metadata: serde_json::Value = serde_json::from_str(
+            request["client_metadata"]["x-codex-turn-metadata"]
+                .as_str()
+                .expect("guardian turn metadata"),
+        )?;
+        assert_eq!(metadata["thread_source"], "guardian_review");
+    }
     assert_eq!(
         guardian_review["client_metadata"]["thread_id"].as_str(),
         Some(guardian_thread_id)
@@ -442,10 +451,18 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review(
         .await
         .expect("guardian trunk rollout path");
     test.codex.shutdown_and_wait().await?;
-    let guardian_context_windows = fs::read_to_string(guardian_rollout_path)?
+    let guardian_rollout = fs::read_to_string(guardian_rollout_path)?
         .lines()
         .map(serde_json::from_str::<RolloutLine>)
-        .collect::<serde_json::Result<Vec<_>>>()?
+        .collect::<serde_json::Result<Vec<_>>>()?;
+    assert_eq!(
+        guardian_rollout.iter().find_map(|line| match &line.item {
+            RolloutItem::SessionMeta(meta) => meta.meta.thread_source.as_ref(),
+            _ => None,
+        }),
+        Some(&ThreadSource::GuardianReview)
+    );
+    let guardian_context_windows = guardian_rollout
         .into_iter()
         .filter_map(|line| match line.item {
             RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => Some(event.model_context_window),

codex-rs/core/tests/suite/subagent_notifications.rs

@@ -977,6 +977,12 @@ async fn spawned_child_receives_forked_parent_context(
     let child_request = wait_for_request_with_model(&child_request_log, REQUESTED_MODEL).await?;
     assert!(child_request.body_contains_text(TURN_0_FORK_PROMPT));
     let child_body = child_request.body_json();
+    let child_metadata: serde_json::Value = serde_json::from_str(
+        child_body["client_metadata"]["x-codex-turn-metadata"]
+            .as_str()
+            .expect("child turn metadata"),
+    )?;
+    assert_eq!(child_metadata["thread_source"], "subagent");
     let original_parent_turn_id = parent_body["client_metadata"]["turn_id"]
         .as_str()
         .expect("legacy spawn parent turn id");

codex-rs/protocol/src/protocol.rs

@@ -2604,6 +2604,7 @@ pub enum SessionSource {
 pub enum ThreadSource {
     User,
     Subagent,
+    GuardianReview,
     Feature(String),
     MemoryConsolidation,
 }
@@ -2613,6 +2614,7 @@ impl ThreadSource {
         match self {
             ThreadSource::User => "user",
             ThreadSource::Subagent => "subagent",
+            ThreadSource::GuardianReview => "guardian_review",
             ThreadSource::Feature(feature) => feature,
             ThreadSource::MemoryConsolidation => "memory_consolidation",
         }
@@ -2646,6 +2648,7 @@ impl FromStr for ThreadSource {
         match value {
             "user" => Ok(ThreadSource::User),
             "subagent" => Ok(ThreadSource::Subagent),
+            "guardian_review" => Ok(ThreadSource::GuardianReview),
             "memory_consolidation" => Ok(ThreadSource::MemoryConsolidation),
             other => Ok(ThreadSource::Feature(other.to_string())),
         }