use anyhow::Result; use codex_core::features::Feature; use core_test_support::responses; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_response_created; use core_test_support::responses::mount_sse_once; use core_test_support::responses::mount_sse_sequence; use core_test_support::responses::sse; use core_test_support::skip_if_no_network; use core_test_support::test_codex::test_codex; use pretty_assertions::assert_eq; use wiremock::Mock; use wiremock::ResponseTemplate; use wiremock::http::Method; use wiremock::matchers::method; use wiremock::matchers::path_regex; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn websocket_fallback_switches_to_http_on_upgrade_required_connect() -> Result<()> { skip_if_no_network!(Ok(())); let server = responses::start_mock_server().await; Mock::given(method("GET")) .and(path_regex(".*/responses$")) .respond_with(ResponseTemplate::new(426)) .mount(&server) .await; let response_mock = mount_sse_once( &server, sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), ) .await; let mut builder = test_codex().with_config({ let base_url = format!("{}/v1", server.uri()); move |config| { config.model_provider.base_url = Some(base_url); config.model_provider.wire_api = codex_core::WireApi::Responses; config.features.enable(Feature::ResponsesWebsockets); // If we don't treat 426 specially, the sampling loop would retry the WebSocket // handshake before switching to the HTTP transport. config.model_provider.stream_max_retries = Some(2); config.model_provider.request_max_retries = Some(0); } }); let test = builder.build(&server).await?; test.submit_turn("hello").await?; let requests = server.received_requests().await.unwrap_or_default(); let websocket_attempts = requests .iter() .filter(|req| req.method == Method::GET && req.url.path().ends_with("/responses")) .count(); let http_attempts = requests .iter() .filter(|req| req.method == Method::POST && req.url.path().ends_with("/responses")) .count(); // One websocket attempt comes from startup preconnect and one from the first turn's stream // attempt before fallback activates; after fallback, transport is HTTP. This matches the // retry-budget tradeoff documented in [`codex_core::client`] module docs. assert_eq!(websocket_attempts, 2); assert_eq!(http_attempts, 1); assert_eq!(response_mock.requests().len(), 1); Ok(()) } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn websocket_fallback_switches_to_http_after_retries_exhausted() -> Result<()> { skip_if_no_network!(Ok(())); let server = responses::start_mock_server().await; let response_mock = mount_sse_once( &server, sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), ) .await; let mut builder = test_codex().with_config({ let base_url = format!("{}/v1", server.uri()); move |config| { config.model_provider.base_url = Some(base_url); config.model_provider.wire_api = codex_core::WireApi::Responses; config.features.enable(Feature::ResponsesWebsockets); config.model_provider.stream_max_retries = Some(2); config.model_provider.request_max_retries = Some(0); } }); let test = builder.build(&server).await?; test.submit_turn("hello").await?; let requests = server.received_requests().await.unwrap_or_default(); let websocket_attempts = requests .iter() .filter(|req| req.method == Method::GET && req.url.path().ends_with("/responses")) .count(); let http_attempts = requests .iter() .filter(|req| req.method == Method::POST && req.url.path().ends_with("/responses")) .count(); // One websocket attempt comes from startup preconnect. // The first turn then makes 3 websocket stream attempts (initial try + 2 retries), // after which fallback activates and the request is replayed over HTTP. assert_eq!(websocket_attempts, 4); assert_eq!(http_attempts, 1); assert_eq!(response_mock.requests().len(), 1); Ok(()) } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn websocket_fallback_is_sticky_across_turns() -> Result<()> { skip_if_no_network!(Ok(())); let server = responses::start_mock_server().await; let response_mock = mount_sse_sequence( &server, vec![ sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), sse(vec![ev_response_created("resp-2"), ev_completed("resp-2")]), ], ) .await; let mut builder = test_codex().with_config({ let base_url = format!("{}/v1", server.uri()); move |config| { config.model_provider.base_url = Some(base_url); config.model_provider.wire_api = codex_core::WireApi::Responses; config.features.enable(Feature::ResponsesWebsockets); config.model_provider.stream_max_retries = Some(2); config.model_provider.request_max_retries = Some(0); } }); let test = builder.build(&server).await?; test.submit_turn("first").await?; test.submit_turn("second").await?; let requests = server.received_requests().await.unwrap_or_default(); let websocket_attempts = requests .iter() .filter(|req| req.method == Method::GET && req.url.path().ends_with("/responses")) .count(); let http_attempts = requests .iter() .filter(|req| req.method == Method::POST && req.url.path().ends_with("/responses")) .count(); // WebSocket attempts all happen on the first turn: // 1 startup preconnect + 3 stream attempts (initial try + 2 retries) before fallback. // Fallback is sticky, so the second turn stays on HTTP and adds no websocket attempts. assert_eq!(websocket_attempts, 4); assert_eq!(http_attempts, 2); assert_eq!(response_mock.requests().len(), 2); Ok(()) }