Emit prompt telemetry only after hooks accept input (#56234)

## Why

Prompt telemetry can include full user text. Emitting it before hook acceptance logs prompts that `UserPromptSubmit` hooks subsequently block.

## What changed

Move prompt telemetry from input submission to hook acceptance, including queued and steered input. Preserve telemetry for accepted Guardian review prompts that are converted to response items during finalization.

## Testing

Extend hook regression tests to assert that blocked prompts are absent from telemetry and rollouts, while accepted queued prompts are logged exactly once. Extend Guardian review coverage to assert exactly one prompt event across inspection steps.

GitOrigin-RevId: aa71f98de2356dd08f62ff492e897ab3a43b54c9

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

@@ -394,6 +394,18 @@ pub(crate) async fn run_turn(
     {
         return Ok(None);
     }
+
+    // Guardian can replace its submitted prompt with response items during finalization.
+    // Those skip UserInput telemetry above, so emit the saved prompt after acceptance.
+    if crate::guardian::is_basic_session_source(&turn_context.session_source)
+        && !user_input.is_empty()
+        && input
+            .iter()
+            .all(|item| matches!(item, TurnInput::ResponseItem(_)))
+    {
+        turn_context.session_telemetry.user_prompt(&user_input);
+    }
+
     // Only speculate after hooks accept the turn, using its finalized tools and permissions.
     {
         let mut state = sess.state.lock().await;
@@ -960,6 +972,10 @@ pub(crate) async fn run_hooks_and_record_inputs(
             blocked_input = true;
             record_additional_contexts(sess, turn_context, hook_outcome.additional_contexts).await;
         } else {
+            if let TurnInput::UserInput { content, .. } = input_item {
+                // Prompt telemetry can contain the full text, so emit it only after hooks accept it.
+                turn_context.session_telemetry.user_prompt(content);
+            }
             if matches!(input_item, TurnInput::UserInput { content, .. } if !content.is_empty()) {
                 accepted_user_input = true;
             }

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

@@ -425,9 +425,6 @@ async fn start_or_steer(
             session
                 .maybe_emit_model_warnings_for_turn(turn_context.as_ref())
                 .await;
-            if let SubmittedTurnInput::UserInput { content, .. } = &input {
-                turn_context.session_telemetry.user_prompt(content);
-            }
             let mut task_input = merge_additional_context_input(session, additional_context).await;
             if has_explicit_input {
                 task_input.push(
@@ -583,9 +580,6 @@ async fn start_if_idle(
     match kind {
         TurnStartKind::User => {
             session.clear_connector_selection().await;
-            if let SubmittedTurnInput::UserInput { content, .. } = &input {
-                turn_context.session_telemetry.user_prompt(content);
-            }
             task_input.push(
                 pending_turn_input(
                     session,
@@ -876,20 +870,14 @@ impl Session {
             .update_responsesapi_client_metadata(responsesapi_client_metadata);
 
         let input = match input {
-            SubmittedTurnInput::UserInput { content, client_id } => {
-                active_task
-                    .turn_context
-                    .session_telemetry
-                    .user_prompt(content);
-                TurnInput::UserInput {
-                    content: std::mem::take(content),
-                    client_id: client_id.clone(),
-                    metadata: super::UserInputMetadata {
-                        acceptance_order: Some(self.reserve_user_input_order().await),
-                        origin,
-                    },
-                }
-            }
+            SubmittedTurnInput::UserInput { content, client_id } => TurnInput::UserInput {
+                content: std::mem::take(content),
+                client_id: client_id.clone(),
+                metadata: super::UserInputMetadata {
+                    acceptance_order: Some(self.reserve_user_input_order().await),
+                    origin,
+                },
+            },
             input => {
                 pending_turn_input(self, input.clone(), sender_messages, active_turn_id, origin)
                     .await

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

@@ -2145,6 +2145,7 @@ async fn guardian_denial_rejects_tool_call_with_rationale(
         })
         .with_config(move |config| {
             config.feedback_enabled = feedback_enabled;
+            config.otel.log_user_prompt = true;
             config.otel.log_guardian_assessments = log_assessments;
             config.otel.exporter = OtelExporterKind::OtlpGrpc {
                 endpoint: "http://127.0.0.1:1".to_string(),
@@ -2367,6 +2368,25 @@ async fn guardian_denial_rejects_tool_call_with_rationale(
             .expect("captured logs")
             .clone(),
     )?;
+    // Finalization converts the accepted review prompt into response items;
+    // it must still emit exactly one prompt event, even across inspection steps.
+    let guardian_thread_id = guardian_request.body_json()["client_metadata"]["thread_id"]
+        .as_str()
+        .expect("Guardian thread ID")
+        .to_owned();
+    let prompt_events: Vec<_> = logs
+        .lines()
+        .filter(|line| {
+            line.contains("codex.user_prompt")
+                && line.contains(" prompt=")
+                && line.contains(&format!("conversation.id={guardian_thread_id}"))
+        })
+        .collect();
+    assert_eq!(
+        prompt_events.len(),
+        1,
+        "Guardian prompt telemetry: {prompt_events:?}"
+    );
     let assessments: Vec<_> = logs
         .lines()
         .filter(|line| {

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

@@ -83,6 +83,7 @@ use tempfile::TempDir;
 use tokio::sync::oneshot;
 use tokio::time::sleep;
 use tokio::time::timeout;
+use tracing_test::traced_test;
 
 const FIRST_CONTINUATION_PROMPT: &str = "Retry with exactly the phrase meow meow meow.";
 const SECOND_CONTINUATION_PROMPT: &str = "Now tighten it to just: meow.";
@@ -2717,7 +2718,9 @@ async fn multiple_blocking_stop_hooks_persist_multiple_hook_prompt_fragments() -
     Ok(())
 }
 
+/// A blocked prompt keeps hook context but never reaches prompt telemetry, the rollout, or the model.
 #[tokio::test]
+#[traced_test]
 async fn blocked_user_prompt_submit_persists_additional_context_for_next_turn() -> Result<()> {
     skip_if_no_network!(Ok(()));
 
@@ -2737,12 +2740,41 @@ async fn blocked_user_prompt_submit_persists_additional_context_for_next_turn()
             write_user_prompt_submit_hook(home, "blocked first prompt", BLOCKED_PROMPT_CONTEXT)
                 .expect("failed to write user prompt submit hook test fixture");
         })
-        .with_config(trust_discovered_hooks);
+        .with_config(|config| {
+            trust_discovered_hooks(config);
+            config.otel.log_user_prompt = true;
+        });
     let test = builder.build(&server).await?;
 
-    test.submit_turn("blocked first prompt").await?;
+    test.codex
+        .start_turn_if_idle(TurnInputRequest::user_input(vec![UserInput::Text {
+            text: "blocked first prompt".to_string(),
+            text_elements: Vec::new(),
+        }]))
+        .await?;
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
     test.submit_turn("second prompt").await?;
 
+    logs_assert(|lines| {
+        let prompts: Vec<_> = lines
+            .iter()
+            .filter(|line| line.contains("codex.user_prompt") && line.contains(" prompt="))
+            .collect();
+        if prompts.len() == 1 && prompts[0].contains("prompt=second prompt") {
+            Ok(())
+        } else {
+            Err(format!(
+                "expected only the accepted prompt in telemetry: {prompts:?}"
+            ))
+        }
+    });
+    let rollout = fs::read_to_string(test.codex.rollout_path().expect("rollout path"))?;
+    assert!(!rollout.contains("blocked first prompt"));
+    assert!(rollout.contains("second prompt"));
+
     let request = response.single_request();
     assert!(
         request
@@ -2792,7 +2824,9 @@ async fn blocked_user_prompt_submit_persists_additional_context_for_next_turn()
     Ok(())
 }
 
+/// Queued input is logged exactly once after acceptance; a later blocked prompt cannot leak or strand it.
 #[tokio::test]
+#[traced_test]
 async fn blocked_queued_prompt_does_not_strand_earlier_accepted_prompt() -> Result<()> {
     skip_if_no_network!(Ok(()));
 
@@ -2838,6 +2872,7 @@ async fn blocked_queued_prompt_does_not_strand_earlier_accepted_prompt() -> Resu
         })
         .with_config(move |config| {
             trust_discovered_hooks(config);
+            config.otel.log_user_prompt = true;
         });
     let test = builder.build_with_streaming_server(&server).await?;
 
@@ -2886,6 +2921,26 @@ async fn blocked_queued_prompt_does_not_strand_earlier_accepted_prompt() -> Resu
 
     assert_eq!(requests.len(), 2);
 
+    logs_assert(|lines| {
+        let prompts: Vec<_> = lines
+            .iter()
+            .filter(|line| line.contains("codex.user_prompt") && line.contains(" prompt="))
+            .collect();
+        if prompts.len() == 2
+            && prompts[0].contains("prompt=initial prompt")
+            && prompts[1].contains("prompt=accepted queued prompt")
+        {
+            Ok(())
+        } else {
+            Err(format!(
+                "expected only accepted prompts in telemetry: {prompts:?}"
+            ))
+        }
+    });
+    let rollout = fs::read_to_string(test.codex.rollout_path().expect("rollout path"))?;
+    assert!(!rollout.contains("blocked queued prompt"));
+    assert!(rollout.contains("accepted queued prompt"));
+
     let second_user_texts = request_message_input_texts(&requests[1], "user");
     assert!(
         second_user_texts.contains(&"accepted queued prompt".to_string()),