Move content-filter guidance into the shared Responses retry handler (#49130)

## What changed

Move content-filter guidance recording from the sampling loop into
`handle_response_stream_error`. Pass `StepContext` so the handler can resolve
model-specific guidance and record it before retry decisions for sampling
requests. Keep remote compaction excluded from guidance recording and update
callers and the existing retry-deadline test to use the new signature.

GitOrigin-RevId: 030bde1d4f1597fc4f3971ad837478e5304c5aa9

codex-rs/core/src/compact_remote_v2.rs

@@ -428,7 +428,7 @@ async fn run_remote_compaction_request_v2(
                     err,
                     client_session,
                     sess,
-                    turn_context,
+                    step_context,
                     ResponsesStreamRequest::RemoteCompactionV2,
                 )
                 .await?;

codex-rs/core/src/responses_retry.rs

@@ -1,9 +1,13 @@
 //! Shared retry and transport fallback decisions for Responses requests.
+//! Content-filter guidance is recorded for sampling requests before retry decisions.
 
 use std::time::Duration;
 
 use crate::client::ModelClientSession;
+use crate::context::ContentFilterGuidance;
+use crate::context::ContextualUserFragment;
 use crate::session::session::Session;
+use crate::session::step_context::StepContext;
 use crate::session::turn_context::TurnContext;
 use codex_client::RetryOperation;
 use codex_features::Feature;
@@ -55,9 +59,26 @@ pub(crate) async fn handle_response_stream_error(
     err: CodexErr,
     client_session: &mut ModelClientSession,
     sess: &Session,
-    turn_context: &TurnContext,
+    step_context: &StepContext,
     request: ResponsesStreamRequest,
 ) -> Result<(), CodexErr> {
+    let turn_context = &step_context.turn;
+    if matches!(request, ResponsesStreamRequest::Sampling)
+        && matches!(err.details(), CodexErrorDetails::ContentFilter)
+    {
+        let model_info = &step_context.settings.model_info;
+        let guidance = ContentFilterGuidance {
+            text: codex_prompts::ResolvedModelMessages::from_model(model_info)
+                .content_filter_guidance()
+                .to_string(),
+        };
+        sess.record_conversation_items(
+            turn_context,
+            model_info,
+            &[ContextualUserFragment::into(guidance)],
+        )
+        .await;
+    }
     let operation = match request {
         ResponsesStreamRequest::Sampling => RetryOperation::Sampling,
         ResponsesStreamRequest::RemoteCompactionV2 => RetryOperation::RemoteCompactionV2,

codex-rs/core/src/responses_retry_tests.rs

@@ -3,9 +3,11 @@ use super::ResponsesStreamRetryState;
 use super::handle_response_stream_error;
 use super::log_retry;
 use crate::realtime_history::RealtimeHistoryState;
+use crate::session::step_context::StepContext;
 use crate::session::tests::make_session_and_context;
 use codex_http_client::RetryAfter;
 use codex_protocol::error::CodexErr;
+use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::Mutex;
 use tokio::time::Instant;
@@ -56,6 +58,7 @@ async fn sampling_retry_logs_stream_error_context() {
 )]
 async fn stream_retry_preserves_deadline_across_delayed_notification() {
     let (mut session, turn_context) = make_session_and_context().await;
+    let step_context = StepContext::for_test(Arc::new(turn_context));
     session.realtime_history = Some(Mutex::new(RealtimeHistoryState::default()));
     let mut client_session = session.services.model_client.new_session();
     // The second retry emits a notification even when release builds hide the first.
@@ -76,7 +79,7 @@ async fn stream_retry_preserves_deadline_across_delayed_notification() {
         CodexErr::InternalServerError.with_retry_after(advice),
         &mut client_session,
         &session,
-        &turn_context,
+        &step_context,
         ResponsesStreamRequest::Sampling,
     );
     tokio::pin!(retry);

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

@@ -11,7 +11,6 @@ use crate::compact::InitialContextInjection;
 use crate::compact::run_inline_auto_compact_task;
 use crate::compact_remote_v2::run_inline_remote_auto_compact_task as run_inline_remote_auto_compact_task_v2;
 use crate::connectors;
-use crate::context::ContentFilterGuidance;
 use crate::context::ContextualUserFragment;
 use crate::context::UserVerificationNotice;
 use crate::environment_selection::TurnEnvironmentSnapshot;
@@ -1698,33 +1697,15 @@ async fn run_sampling_request(
 
         let original_input = original_input.get_or_insert(prompt.input);
 
-        let retry = async {
-            if matches!(err.details(), CodexErrorDetails::ContentFilter) {
-                let model_info = &step_context.settings.model_info;
-                let guidance = ContentFilterGuidance {
-                    text: codex_prompts::ResolvedModelMessages::from_model(model_info)
-                        .content_filter_guidance()
-                        .to_string(),
-                };
-                sess.record_conversation_items(
-                    &turn_context,
-                    model_info,
-                    &[ContextualUserFragment::into(guidance)],
-                )
-                .await;
-            }
-
-            handle_response_stream_error(
-                &mut retry_state,
-                max_retries,
-                err,
-                client_session,
-                &sess,
-                &turn_context,
-                ResponsesStreamRequest::Sampling,
-            )
-            .await
-        }
+        let retry = handle_response_stream_error(
+            &mut retry_state,
+            max_retries,
+            err,
+            client_session,
+            &sess,
+            &step_context,
+            ResponsesStreamRequest::Sampling,
+        )
         .or_cancel(&preempt)
         .or_cancel(&cancellation_token)
         .await?;