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()),