Treat content policy violations as terminal errors (#60479)

## What changed

Recognize `content_policy_violation` in HTTP 400 responses and streamed errors as a dedicated, non-retryable error. Preserve the provider's message and available HTTP status, use a fallback message when needed, and report `CodexErrorInfo::InvalidPrompt` to clients.

End the response stream immediately on this error and stop compaction from falling back to the current model after a content policy rejection. Add redacted diagnostic output and a dedicated sampler failure reason.

## Testing

Extend coverage for HTTP and wrapped WebSocket errors, missing or blank provider messages, terminal stream failures with retry headers, and content policy rejections during local and remote compaction on model downshift.

GitOrigin-RevId: 592170ac30c07b0b303a8d5cb15bd35c8f821b55

codex-rs/codex-api/src/api_bridge.rs

@@ -43,6 +43,7 @@ pub fn map_api_error(err: ApiError) -> CodexErr {
         | ApiError::InvalidAudioInput { .. }
         | ApiError::ModelNotFound { .. }
         | ApiError::InvalidPrompt { .. }
+        | ApiError::ContentPolicyViolation { .. }
         | ApiError::CyberPolicy { .. }
         | ApiError::BioPolicy { .. }
         | ApiError::MisalignmentPolicyViolation { .. } => None,
@@ -104,6 +105,9 @@ fn map_api_error_details(err: ApiError) -> CodexErr {
         ApiError::InvalidPrompt { message } => {
             CodexErr::new(CodexErrorDetails::InvalidPrompt { message })
         }
+        ApiError::ContentPolicyViolation { message, status } => {
+            CodexErrorDetails::ContentPolicyViolation { message, status }.into()
+        }
         ApiError::CyberPolicy { message } => {
             CodexErr::new(CodexErrorDetails::CyberPolicy { message })
         }

codex-rs/codex-api/src/api_bridge_tests.rs

@@ -358,6 +358,11 @@ fn map_api_error_preserves_typed_errors() {
 #[test]
 fn map_api_error_maps_http_and_wrapped_websocket_typed_errors() {
     for (code, expected_info, fallback) in [
+        (
+            "content_policy_violation",
+            CodexErrorInfo::InvalidPrompt,
+            "The input was rejected by the content policy.",
+        ),
         (
             "bio_policy",
             CodexErrorInfo::BioPolicy,

codex-rs/codex-api/src/error.rs

@@ -56,6 +56,11 @@ pub enum ApiError {
     },
     #[error("invalid prompt: {message}")]
     InvalidPrompt { message: String },
+    #[error("{message}")]
+    ContentPolicyViolation {
+        message: String,
+        status: Option<StatusCode>,
+    },
     #[error("cyber policy: {message}")]
     CyberPolicy { message: String },
     #[error("bio policy: {message}")]
@@ -87,6 +92,17 @@ pub(crate) fn parse_terminal_provider_error(
     let error = error?;
     let code = error.get("code")?.as_str()?;
     match (code, status) {
+        ("content_policy_violation", None | Some(StatusCode::BAD_REQUEST)) => {
+            Some(ApiError::ContentPolicyViolation {
+                message: error
+                    .get("message")
+                    .and_then(Value::as_str)
+                    .filter(|message| !message.trim().is_empty())
+                    .unwrap_or("The input was rejected by the content policy.")
+                    .to_owned(),
+                status,
+            })
+        }
         // HTTP 403s retain their request ID and headers in the API bridge.
         ("access_program_not_enabled", None) => Some(ApiError::AccessProgramNotEnabled {
             message: error

codex-rs/codex-api/src/sse/responses.rs

@@ -682,6 +682,7 @@ async fn process_sse_with_treatment(
                     ApiError::FlexUnavailable
                         | ApiError::SessionBudgetExceeded
                         | ApiError::AccessProgramNotEnabled { .. }
+                        | ApiError::ContentPolicyViolation { .. }
                         | ApiError::ResponseIncomplete(_)
                 ) {
                     let _ = tx_event.send(Err(error)).await;

codex-rs/codex-api/src/sse/responses_error_tests.rs

@@ -115,6 +115,7 @@ async fn terminal_failure_with_retry_header_ends_stream_immediately() {
     for (code, expected) in [
         ("flex_unavailable", CodexErrorInfo::FlexUnavailable),
         ("budget_reached", CodexErrorInfo::SessionBudgetExceeded),
+        ("content_policy_violation", CodexErrorInfo::InvalidPrompt),
         (
             "access_program_not_enabled",
             CodexErrorInfo::HttpConnectionFailed {
@@ -241,6 +242,7 @@ async fn streamed_retry_after_uses_http_header_validation() {
 async fn retry_header_survives_other_failed_responses() {
     for (code, expected_code) in [
         ("server_is_overloaded", CodexErrorInfo::ServerOverloaded),
+        ("safety_violation", CodexErrorInfo::Other),
         ("unknown_failure", CodexErrorInfo::Other),
     ] {
         let event = serde_json::from_value(json!({