From 8c7affbe1b8a912e92dbfd75b8d0cceb9ca34507 Mon Sep 17 00:00:00 2001 From: Abhinav Date: Thu, 22 Oct 2026 17:36:05 +0000 Subject: [PATCH] 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 | 16 ++++++ codex-rs/core/src/session/turn_input.rs | 28 +++------- codex-rs/core/tests/suite/guardian_review.rs | 20 +++++++ codex-rs/core/tests/suite/hooks.rs | 59 +++++++++++++++++++- 4 files changed, 101 insertions(+), 22 deletions(-) diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 8531b098ba..65cbd64a18 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/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; } diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index 4d2f83bbe0..793cdf3da5 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/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 diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index c18040e798..ec0ac13dc6 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/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| { diff --git a/codex-rs/core/tests/suite/hooks.rs b/codex-rs/core/tests/suite/hooks.rs index 9fce5b9c3d..1f76244051 100644 --- a/codex-rs/core/tests/suite/hooks.rs +++ b/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()),