Stop retrying explicit invalid-workspace errors (#60475)

## Why

HTTP 403 responses with `invalid_workspace_selected` were treated as unexpected status errors, allowing retries and compaction fallback to another model after the server rejected the selected workspace.

## What changed

Recognize the code in `detail.code` or `error.code`, falling back to the `x-error-json` header when the body has no code. Classify these 403 responses as `InvalidWorkspaceSelected` and disable retries and compaction model fallback while preserving the response details and HTTP status. Other 403 responses retain their existing retry behavior.

## Testing

Add coverage for body and header error codes, code precedence, HTTP and WebSocket retry behavior, and local and remote compaction during a model downshift.

GitOrigin-RevId: f44a3bee5ef06e601909839868ab99cea6dd7d03

codex-rs/core/src/compact_model_fallback.rs

@@ -17,6 +17,7 @@ pub(crate) fn should_retry_with_current_model(
         CodexErrorDetails::TurnAborted
             | CodexErrorDetails::Interrupted
             | CodexErrorDetails::SessionBudgetExceeded
+            | CodexErrorDetails::InvalidWorkspaceSelected(_)
             | CodexErrorDetails::ResponseIncomplete(_)
     ) || (previous.model_info().slug == current.model_info().slug
         && previous.cyber_access_program == current.cyber_access_program)

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

@@ -1140,6 +1140,63 @@ async fn reasoning_effort_override_remote_v2_compaction_resets_pinned_effort(
     Ok(())
 }
 
+#[test_case::test_case(non_openai_model_provider; "local")]
+#[test_case::test_case(openai_model_provider; "remote")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn workspace_denial_stops_compaction_on_model_downshift(
+    provider: fn(&MockServer) -> ModelProviderInfo,
+) -> Result<()> {
+    skip_if_no_network!(Ok(()));
+    let server = start_mock_server().await;
+    let previous = "gpt-5.4";
+    let selected = "gpt-5.2";
+    let model_provider = provider(&server);
+    let test = test_codex()
+        .with_model(previous)
+        .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
+        .with_config(move |config| {
+            config.model_provider = model_provider;
+            config.model_provider.stream_max_retries = Some(2);
+            config.model_catalog = Some(ModelsResponse {
+                models: vec![
+                    model_info_with_context_window(previous, /*context_window*/ 273_000),
+                    model_info_with_context_window(selected, /*context_window*/ 125_000),
+                ],
+            });
+            set_test_compact_prompt(config);
+        })
+        .build_with_auto_env(&server)
+        .await?;
+    mount_sse_once(
+        &server,
+        sse(vec![
+            ev_assistant_message("message", "First reply"),
+            ev_completed_with_tokens("first", /*total_tokens*/ 120_000),
+        ]),
+    )
+    .await;
+    test.submit_turn("first turn").await?;
+
+    let denied = responses::mount_response_handler(&server, |_| {
+        wiremock::ResponseTemplate::new(403)
+            .set_body_json(json!({ "detail": { "code": "invalid_workspace_selected" } }))
+    })
+    .await;
+    test.codex
+        .start_or_steer_turn(disabled_permission_user_turn(
+            "continue on the smaller model",
+            test.cwd.path().to_path_buf(),
+            selected.to_string(),
+        ))
+        .await?;
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+    assert_eq!(denied.single_request().body_json()["model"], previous);
+    Ok(())
+}
+
 #[test_case::test_case(json!({"input_tokens": 120_000, "output_tokens": 30_000, "total_tokens": 150_000}); "reported_usage")]
 #[test_case::test_case(Value::Null; "missing_usage")]
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]

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

@@ -127,6 +127,49 @@ async fn websocket_fallback_switches_to_http_after_retries_exhausted() -> Result
     Ok(())
 }
 
+#[test_case::test_case(true, Some("invalid_workspace_selected"), 3, 0; "workspace denial on websocket")]
+#[test_case::test_case(false, Some("invalid_workspace_selected"), 0, 2; "workspace denial on http")]
+#[test_case::test_case(true, None, 3, 4; "bare websocket and http denials")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn forbidden_responses_retry_only_without_workspace_denial(
+    supports_websockets: bool,
+    code: Option<&str>,
+    expected_upgrades: u64,
+    expected_posts: u64,
+) -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let server = responses::start_mock_server().await;
+    let mut denial = ResponseTemplate::new(403);
+    if let Some(code) = code {
+        denial = denial.set_body_json(serde_json::json!({"detail": {"code": code}}));
+    }
+    for (verb, count) in [("GET", expected_upgrades), ("POST", expected_posts)] {
+        Mock::given(method(verb))
+            .and(path_regex(".*/responses$"))
+            .respond_with(denial.clone())
+            .expect(count)
+            .mount(&server)
+            .await;
+    }
+
+    let test = test_codex()
+        .with_config(move |config| {
+            config.model_provider.supports_websockets = supports_websockets;
+            config.model_provider.stream_max_retries = Some(1);
+            config.model_provider.request_max_retries = Some(1);
+        })
+        .build_with_auto_env(&server)
+        .await?;
+
+    // A WebSocket-enabled session also attempts a startup prewarm. A typed denial
+    // should leave the next turn eligible for its own WebSocket attempt.
+    test.submit_turn("first").await?;
+    test.submit_turn("second").await?;
+    server.verify().await;
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn websocket_fallback_hides_first_websocket_retry_stream_error() -> Result<()> {
     skip_if_no_network!(Ok(()));