diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock
index 62e943d0f42..a72a55d8239 100644
--- a/litellm-rust/Cargo.lock
+++ b/litellm-rust/Cargo.lock
@@ -1469,6 +1469,7 @@ dependencies = [
  "litellm-python-interop",
  "pyo3",
  "pyo3-async-runtimes",
+ "rstest",
  "serde",
  "serde_json",
  "tokio",
diff --git a/litellm-rust/crates/ai-gateway/src/audio_transcription/hooks.rs b/litellm-rust/crates/ai-gateway/src/audio_transcription/hooks.rs
index dbe2d3a325b..c1c7b911a08 100644
--- a/litellm-rust/crates/ai-gateway/src/audio_transcription/hooks.rs
+++ b/litellm-rust/crates/ai-gateway/src/audio_transcription/hooks.rs
@@ -269,6 +269,7 @@ fn guardrail_error_to_core_error(error: GuardrailError) -> Error {
 
 fn core_error_kind(error: &Error) -> &'static str {
     match error {
+        Error::Declined(_) => "UnsupportedRequest",
         Error::Auth(_) => "AuthError",
         Error::InvalidProvider(_) => "InvalidProvider",
         Error::InvalidRequest(_) => "InvalidRequest",
diff --git a/litellm-rust/crates/ai-gateway/src/ocr/hooks.rs b/litellm-rust/crates/ai-gateway/src/ocr/hooks.rs
index 446b323db3a..9103b942684 100644
--- a/litellm-rust/crates/ai-gateway/src/ocr/hooks.rs
+++ b/litellm-rust/crates/ai-gateway/src/ocr/hooks.rs
@@ -374,6 +374,7 @@ fn guardrail_error_to_core_error(error: GuardrailError) -> Error {
 
 fn core_error_kind(error: &Error) -> &'static str {
     match error {
+        Error::Declined(_) => "UnsupportedRequest",
         Error::Auth(_) => "AuthError",
         Error::InvalidProvider(_) => "InvalidProvider",
         Error::InvalidRequest(_) => "InvalidRequest",
diff --git a/litellm-rust/crates/ai-gateway/src/routes/messages/mod.rs b/litellm-rust/crates/ai-gateway/src/routes/messages/mod.rs
index bb9f3851a77..b45212f0af8 100644
--- a/litellm-rust/crates/ai-gateway/src/routes/messages/mod.rs
+++ b/litellm-rust/crates/ai-gateway/src/routes/messages/mod.rs
@@ -121,7 +121,7 @@ impl IntoResponse for MessagesRouteError {
             // The gateway has no Python implementation to decline to, so a
             // request the core cannot serve is reported to the caller. The
             // reason is a fixed internal string, never provider content.
-            Error::Unsupported(reason) => (
+            Error::Declined(reason) | Error::Unsupported(reason) => (
                 StatusCode::BAD_REQUEST,
                 format!("messages request is not supported: {reason}"),
             ),
diff --git a/litellm-rust/crates/core/src/chat_completions/handler.rs b/litellm-rust/crates/core/src/chat_completions/handler.rs
index 96d001e2892..a70d1b78a3a 100644
--- a/litellm-rust/crates/core/src/chat_completions/handler.rs
+++ b/litellm-rust/crates/core/src/chat_completions/handler.rs
@@ -32,9 +32,6 @@ pub(super) async fn execute_chat_completions_provider_call(
     }
 
     let response = http_request(request_builder).await.map_err(|err| {
-        // Failing to establish the connection means the request never went out,
-        // so the host can still serve it. Everything else here, a timeout
-        // above all, may have reached the provider and been answered.
         if err.is_connect() || err.is_builder() {
             Error::Connect(err.to_string())
         } else {
@@ -64,15 +61,6 @@ pub(super) async fn execute_chat_completions_provider_call(
         .map_err(as_response_error)
 }
 
-/// Re-tag an error raised while normalizing a response the provider already
-/// returned.
-///
-/// A config reports the same variants on either side of the call: a missing
-/// field or an unsupported block can mean "this request cannot be translated"
-/// during prepare and "this response cannot be normalized" here. Only the
-/// second kind has already been billed, and a host that keeps a reference
-/// implementation must not retry those, so collapse them to one variant that
-/// can only mean the provider was already called.
 pub(super) fn as_response_error(err: Error) -> Error {
     match err {
         already @ (Error::InvalidResponse(_) | Error::Http { .. }) => already,
diff --git a/litellm-rust/crates/core/src/chat_completions/prepare.rs b/litellm-rust/crates/core/src/chat_completions/prepare.rs
index 3be2ba21de4..dee150e0971 100644
--- a/litellm-rust/crates/core/src/chat_completions/prepare.rs
+++ b/litellm-rust/crates/core/src/chat_completions/prepare.rs
@@ -40,15 +40,15 @@ pub(super) fn parse_messages(messages: Value) -> Result<Vec<ChatMessage>, Error>
 pub(super) fn resolve_request(
     request: ChatCompletionsRequest<'_>,
 ) -> Result<ResolvedChatCompletionsRequest<'_>, Error> {
-    let (model, config) = resolve_provider_config(request.model, request.custom_llm_provider)?;
-    let messages = parse_messages(request.messages)?;
+    let (model, config) = resolve_provider_config(request.model, request.custom_llm_provider)
+        .map_err(|_| Error::Declined("provider is not on the rust chat completions path"))?;
+    let messages =
+        parse_messages(request.messages).map_err(|_| Error::Declined("unreadable message list"))?;
     if messages.is_empty() {
-        return Err(Error::InvalidRequest(
-            "chat completions requires at least one message".to_string(),
-        ));
+        return Err(Error::Declined("empty message list"));
     }
     if let Some(reason) = config.unsupported_reason(&messages, &request.optional_params) {
-        return Err(Error::Unsupported(reason.0));
+        return Err(Error::Declined(reason.0));
     }
     Ok(ResolvedChatCompletionsRequest {
         model,
diff --git a/litellm-rust/crates/core/src/chat_completions/tests.rs b/litellm-rust/crates/core/src/chat_completions/tests.rs
index f8594dee447..0104729650d 100644
--- a/litellm-rust/crates/core/src/chat_completions/tests.rs
+++ b/litellm-rust/crates/core/src/chat_completions/tests.rs
@@ -202,7 +202,7 @@ fn declines_an_unsupported_request_before_resolving_credentials() {
     call.api_key = None;
     // No api_key is set and no env is consulted: the gate must run first, so the
     // error is the decline rather than a missing-credential error.
-    assert_eq!(decline(call), Error::Unsupported("streaming"));
+    assert_eq!(decline(call), Error::Declined("streaming"));
 }
 
 #[test]
@@ -214,21 +214,21 @@ fn rejects_an_unknown_provider() {
             json!([{"role": "user", "content": "hi"}]),
             json!({}),
         )),
-        Error::InvalidProvider("openai".to_string())
+        Error::Declined("provider is not on the rust chat completions path")
     );
 }
 
 #[test]
 fn rejects_a_model_with_no_resolvable_provider() {
-    assert!(matches!(
+    assert_eq!(
         decline(request(
             "claude-sonnet-4-5",
             None,
             json!([{"role": "user", "content": "hi"}]),
             json!({}),
         )),
-        Error::InvalidProvider(_)
-    ));
+        Error::Declined("provider is not on the rust chat completions path")
+    );
 }
 
 #[test]
@@ -240,17 +240,17 @@ fn rejects_an_empty_or_malformed_message_list() {
             json!([]),
             json!({}),
         )),
-        Error::InvalidRequest("chat completions requires at least one message".to_string())
+        Error::Declined("empty message list")
     );
-    assert!(matches!(
+    assert_eq!(
         decline(request(
             "anthropic/claude-sonnet-4-5",
             None,
             json!("not a list"),
             json!({}),
         )),
-        Error::InvalidRequest(_)
-    ));
+        Error::Declined("unreadable message list")
+    );
 }
 
 #[test]
@@ -775,11 +775,7 @@ mod round_trip {
     }
 
     #[tokio::test]
-    async fn a_connection_that_is_never_established_declines_instead_of_failing() {
-        // Nothing was sent, so nothing was billed and the host can still serve
-        // the request. Classing this with the post-send failures would turn a
-        // recoverable fallback into a user-facing error on exactly the
-        // deployments whose transport is configured only on the Python client.
+    async fn a_connection_that_is_never_established_is_terminal() {
         let port = {
             let listener = TcpListener::bind("127.0.0.1:0").await.expect("binds");
             listener.local_addr().expect("has an address").port()
@@ -814,7 +810,6 @@ mod round_trip {
                 "{label} must not stay retryable once the provider has answered"
             );
         }
-        // An upstream status is already unambiguous, so it survives intact.
         assert!(matches!(
             as_response_error(Error::Http {
                 status: 500,
diff --git a/litellm-rust/crates/core/src/error.rs b/litellm-rust/crates/core/src/error.rs
index db3fa2ec704..44075ccfe88 100644
--- a/litellm-rust/crates/core/src/error.rs
+++ b/litellm-rust/crates/core/src/error.rs
@@ -2,6 +2,8 @@ use thiserror::Error as ThisError;
 
 #[derive(Debug, ThisError, PartialEq, Eq)]
 pub enum Error {
+    #[error("native execution declined: {0}")]
+    Declined(&'static str),
     #[error("expected {expected}, got {actual}")]
     InvalidType {
         expected: &'static str,
@@ -21,17 +23,10 @@ pub enum Error {
     Http { status: u16, body: String },
     #[error("upstream network error: {0}")]
     Network(String),
-    /// The provider was never reached: DNS, TCP, TLS or proxy setup failed
-    /// before any byte of the request went out. Nothing was billed, so a host
-    /// that keeps a reference implementation can serve the request itself.
-    /// A timeout is deliberately not this, since the provider may have received
-    /// and answered the request already.
     #[error("could not reach the provider: {0}")]
     Connect(String),
     #[error("routing error: {0}")]
     Routing(String),
-    /// The request is outside the surface this route covers in Rust. Hosts that
-    /// keep a reference implementation treat this as "fall back", not "fail".
     #[error("unsupported by the rust path: {0}")]
     Unsupported(&'static str),
 }
diff --git a/litellm-rust/crates/python-bridge/Cargo.toml b/litellm-rust/crates/python-bridge/Cargo.toml
index bda09a7d840..014346ac5a9 100644
--- a/litellm-rust/crates/python-bridge/Cargo.toml
+++ b/litellm-rust/crates/python-bridge/Cargo.toml
@@ -34,6 +34,7 @@ tokio.workspace = true
 
 [dev-dependencies]
 criterion = "0.8.2"
+rstest.workspace = true
 tokio-tungstenite.workspace = true
 tracing.workspace = true
 
diff --git a/litellm-rust/crates/python-bridge/src/errors.rs b/litellm-rust/crates/python-bridge/src/errors.rs
index 76c298abf89..3ecc4e6e452 100644
--- a/litellm-rust/crates/python-bridge/src/errors.rs
+++ b/litellm-rust/crates/python-bridge/src/errors.rs
@@ -18,6 +18,7 @@ pyo3::create_exception!(
 
 pub(crate) fn core_error_to_pyerr(err: Error) -> PyErr {
     match err {
+        Error::Declined(message) => RustBridgeDeclined::new_err(message),
         Error::Auth(message) => PyValueError::new_err(message),
         Error::InvalidProvider(_)
         | Error::InvalidRequest(_)
@@ -27,30 +28,17 @@ pub(crate) fn core_error_to_pyerr(err: Error) -> PyErr {
     }
 }
 
-/// Map a core error for a route whose host keeps a Python implementation.
-///
-/// The distinction the host needs is whether the provider was already called.
-/// Everything raised before the request goes out is safe for the host to retry
-/// on its own path; anything after it is not, because the provider has already
-/// done the work and billed for it.
 pub(crate) fn chat_completions_error_to_pyerr(err: Error) -> PyErr {
     match err {
-        Error::Unsupported(_)
-        | Error::Auth(_)
-        | Error::InvalidProvider(_)
-        | Error::InvalidRequest(_)
-        | Error::InvalidType { .. }
-        | Error::MissingField(_)
-        | Error::Routing(_)
-        // Nothing reached the provider, so serving it on Python cannot double
-        // bill and is the only way the caller gets an answer at all.
-        | Error::Connect(_) => RustBridgeDeclined::new_err(err.to_string()),
+        Error::Declined(message) => RustBridgeDeclined::new_err(message),
+        Error::Auth(message) => RustUpstreamError::new_err((401u16, message)),
         Error::Http { status, body } => {
             RustUpstreamError::new_err((status, format!("{status}: {body}")))
         }
-        Error::Network(message) | Error::InvalidResponse(message) => {
+        Error::Connect(message) | Error::Network(message) | Error::InvalidResponse(message) => {
             RustUpstreamError::new_err((0u16, message))
         }
+        other => core_error_to_pyerr(other),
     }
 }
 
@@ -71,18 +59,99 @@ pub(crate) fn ocr_error_to_pyerr(err: Error) -> PyErr {
 }
 
 #[cfg(test)]
-mod ocr_error_tests {
+mod tests {
+    use rstest::{fixture, rstest};
+
     use super::*;
 
-    #[test]
-    fn ocr_errors_preserve_python_validation_and_provider_details() {
+    #[fixture]
+    #[once]
+    fn initialized_python() {
         Python::initialize();
+    }
+
+    #[rstest]
+    #[case::connect(Error::Connect, "connection failed")]
+    #[case::network(Error::Network, "connection reset")]
+    #[case::invalid_response(Error::InvalidResponse, "invalid response")]
+    fn chat_transport_errors_do_not_authorize_python_fallback(
+        #[from(initialized_python)] (): (),
+        #[case] error: fn(String) -> Error,
+        #[case] message: &str,
+    ) {
         Python::attach(|py| {
-            for field in ["document_url", "image_url"] {
-                let mapped = ocr_error_to_pyerr(Error::MissingField(field));
-                assert!(mapped.is_instance_of::<PyValueError>(py));
-                assert_eq!(mapped.value(py).to_string(), "Document URL is required");
+            let mapped = chat_completions_error_to_pyerr(error(message.to_string()));
+            assert!(mapped.is_instance_of::<RustUpstreamError>(py));
+            let args: (u16, String) = mapped
+                .value(py)
+                .getattr("args")
+                .and_then(|args| args.extract())
+                .expect("transport errors retain their status and message");
+            assert_eq!(args, (0, message.to_string()));
+        });
+    }
+
+    #[rstest]
+    fn only_explicit_decline_authorizes_python_fallback(#[from(initialized_python)] (): ()) {
+        Python::attach(|py| {
+            let mapped = chat_completions_error_to_pyerr(Error::Declined("unsupported request"));
+            assert!(mapped.is_instance_of::<RustBridgeDeclined>(py));
+            assert_eq!(mapped.value(py).to_string(), "unsupported request");
+        });
+    }
+
+    #[rstest]
+    fn request_failures_do_not_authorize_python_fallback(#[from(initialized_python)] (): ()) {
+        Python::attach(|py| {
+            for error in [
+                Error::InvalidProvider("provider".to_string()),
+                Error::InvalidRequest("invalid".to_string()),
+                Error::Unsupported("unsupported"),
+                Error::Routing("route".to_string()),
+            ] {
+                let mapped = chat_completions_error_to_pyerr(error);
+                assert!(!mapped.is_instance_of::<RustBridgeDeclined>(py));
+                assert!(!mapped.is_instance_of::<RustUpstreamError>(py));
             }
+        });
+    }
+
+    #[rstest]
+    fn credential_failures_do_not_authorize_python_fallback(#[from(initialized_python)] (): ()) {
+        Python::attach(|py| {
+            let mapped = chat_completions_error_to_pyerr(Error::Auth(
+                "credential exchange failed".to_string(),
+            ));
+            assert!(mapped.is_instance_of::<RustUpstreamError>(py));
+            assert_eq!(
+                mapped
+                    .value(py)
+                    .getattr("args")
+                    .unwrap()
+                    .extract::<(u16, String)>()
+                    .unwrap(),
+                (401, "credential exchange failed".to_string())
+            );
+        });
+    }
+
+    #[rstest]
+    #[case::document_url("document_url")]
+    #[case::image_url("image_url")]
+    fn ocr_errors_preserve_python_validation(
+        #[from(initialized_python)] (): (),
+        #[case] field: &'static str,
+    ) {
+        Python::attach(|py| {
+            let mapped = ocr_error_to_pyerr(Error::MissingField(field));
+            assert!(mapped.is_instance_of::<PyValueError>(py));
+            assert_eq!(mapped.value(py).to_string(), "Document URL is required");
+        });
+    }
+
+    #[rstest]
+    fn ocr_errors_preserve_provider_details(#[from(initialized_python)] (): ()) {
+        Python::attach(|py| {
             let mapped = ocr_error_to_pyerr(Error::Http {
                 status: 429,
                 body: r#"{"message":"rate limited"}"#.to_string(),
diff --git a/litellm/llms/anthropic/chat/handler.py b/litellm/llms/anthropic/chat/handler.py
index dc3a4d179b7..bcc2235e11e 100644
--- a/litellm/llms/anthropic/chat/handler.py
+++ b/litellm/llms/anthropic/chat/handler.py
@@ -27,7 +27,7 @@ from litellm.llms.custom_httpx.http_handler import (
 )
 from litellm.rust_bridge import chat_completions as rust_chat_completions_bridge
 from litellm.rust_bridge.chat_completions import rust_chat_completions_accepts
-from litellm.rust_bridge.dispatch import anative_first, native_first
+from litellm.rust_bridge.dispatch import anative_first, native_first, provider_errors
 from litellm.rust_bridge.runtime import DispatchResult
 from litellm.types.llms.anthropic import (
     ContentBlockDelta,
@@ -468,7 +468,7 @@ class AnthropicChatCompletion(BaseLLM):
         @anative_first(
             native=native_acompletion,
             route="chat_completions",
-            errors=lambda: rust_chat_completions_bridge.error_handling(custom_llm_provider or "", model),
+            errors=lambda: provider_errors(custom_llm_provider or "", model),
         )
         async def execute_async() -> ModelResponse | CustomStreamWrapper:
             headers, data = prepare_python()
@@ -525,7 +525,7 @@ class AnthropicChatCompletion(BaseLLM):
         @native_first(
             native=native_completion,
             route="chat_completions",
-            errors=lambda: rust_chat_completions_bridge.error_handling(custom_llm_provider or "", model),
+            errors=lambda: provider_errors(custom_llm_provider or "", model),
         )
         def execute_sync() -> ModelResponse | CustomStreamWrapper:
             headers, data = prepare_python()
diff --git a/litellm/llms/bedrock/audio_transcription/__init__.py b/litellm/llms/bedrock/audio_transcription/__init__.py
index 6d6be069f99..6fc1a82d2c6 100644
--- a/litellm/llms/bedrock/audio_transcription/__init__.py
+++ b/litellm/llms/bedrock/audio_transcription/__init__.py
@@ -5,7 +5,7 @@ import httpx
 
 from litellm.litellm_core_utils.audio_utils.utils import process_audio_file
 from litellm.rust_bridge import transcription as rust_transcription_bridge
-from litellm.rust_bridge.dispatch import PROPAGATE, anative_first, native_first
+from litellm.rust_bridge.dispatch import anative_first, native_first, provider_errors
 from litellm.rust_bridge.runtime import DispatchResult, adapt_result
 from litellm.types.utils import FileTypes, TranscriptionResponse
 
@@ -69,7 +69,7 @@ class BedrockAudioTranscriptionRustDispatch:
         native=_attempt_audio_transcriptions,
         route="audio transcription",
         errors=lambda self, model, audio_file, api_key, api_base, custom_llm_provider, extra_headers, optional_params, timeout: (
-            PROPAGATE
+            provider_errors(custom_llm_provider, model)
         ),
     )
     def audio_transcriptions(
@@ -114,7 +114,7 @@ class BedrockAudioTranscriptionRustDispatch:
         native=_attempt_async_audio_transcriptions,
         route="audio transcription",
         errors=lambda self, model, audio_file, api_key, api_base, custom_llm_provider, extra_headers, optional_params, timeout: (
-            PROPAGATE
+            provider_errors(custom_llm_provider, model)
         ),
     )
     async def async_audio_transcriptions(
diff --git a/litellm/llms/bedrock/chat/converse_handler.py b/litellm/llms/bedrock/chat/converse_handler.py
index 2918ea96b8c..8d6cf23f47e 100644
--- a/litellm/llms/bedrock/chat/converse_handler.py
+++ b/litellm/llms/bedrock/chat/converse_handler.py
@@ -18,7 +18,7 @@ from litellm.llms.custom_httpx.http_handler import (
 )
 from litellm.rust_bridge import chat_completions as rust_chat_completions_bridge
 from litellm.rust_bridge.chat_completions import rust_chat_completions_accepts
-from litellm.rust_bridge.dispatch import anative_first, native_first
+from litellm.rust_bridge.dispatch import anative_first, native_first, provider_errors
 from litellm.rust_bridge.runtime import DispatchResult
 from litellm.types.utils import ModelResponse
 from litellm.utils import CustomStreamWrapper
@@ -458,7 +458,7 @@ class BedrockConverseLLM(BaseAWSLLM):
         @anative_first(
             native=native_acompletion,
             route="chat_completions",
-            errors=lambda: rust_chat_completions_bridge.error_handling("bedrock", model),
+            errors=lambda: provider_errors("bedrock", model),
         )
         async def execute_async() -> ModelResponse | CustomStreamWrapper:
             python_client: Final = None if isinstance(client, HTTPHandler) else client
@@ -506,7 +506,7 @@ class BedrockConverseLLM(BaseAWSLLM):
         @native_first(
             native=native_completion,
             route="chat_completions",
-            errors=lambda: rust_chat_completions_bridge.error_handling("bedrock", model),
+            errors=lambda: provider_errors("bedrock", model),
         )
         def execute_sync() -> ModelResponse | CustomStreamWrapper:
             ## TRANSFORMATION ##
diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py
index 4d6a427e00e..ee5a49374af 100644
--- a/litellm/llms/custom_httpx/llm_http_handler.py
+++ b/litellm/llms/custom_httpx/llm_http_handler.py
@@ -92,7 +92,7 @@ from litellm.responses.streaming_iterator import (
     ResponsesWebSocketStreaming,
     SyncResponsesAPIStreamingIterator,
 )
-from litellm.rust_bridge.dispatch import PYTHON_ON_ERROR, anative_context, anative_first
+from litellm.rust_bridge.dispatch import anative_context, anative_first, provider_errors
 from litellm.rust_bridge.runtime import DispatchResult, NativeSkipped, NativeSkipReason, adapt_result
 from litellm.types.containers.main import (
     ContainerFileListResponse,
@@ -2245,7 +2245,9 @@ class BaseLLMHTTPHandler:
             )
             return adapt_result(result, self._rust_anthropic_messages_fake_stream) if stream else result
 
-        @anative_first(native=native_messages, route="messages", errors=lambda: PYTHON_ON_ERROR)
+        @anative_first(
+            native=native_messages, route="messages", errors=lambda: provider_errors(custom_llm_provider, model)
+        )
         async def execute_messages() -> AnthropicMessagesResponse | AsyncIterator[object]:
             response: Final = await self._async_post_anthropic_messages_with_http_error_retry(
                 async_httpx_client=async_httpx_client,
@@ -6512,7 +6514,11 @@ class BaseLLMHTTPHandler:
                     timeout=timeout,
                 )
 
-            @anative_context(native=attempt_connection, route="responses_websocket", errors=lambda: PYTHON_ON_ERROR)
+            @anative_context(
+                native=attempt_connection,
+                route="responses_websocket",
+                errors=lambda: provider_errors("openai", "responses websocket"),
+            )
             @asynccontextmanager
             async def _backend_connection() -> AsyncGenerator[ClientConnection, None]:
                 async with websockets.connect(
diff --git a/litellm/ocr/main.py b/litellm/ocr/main.py
index 7a3209b9eb4..7fea86e10b8 100644
--- a/litellm/ocr/main.py
+++ b/litellm/ocr/main.py
@@ -25,7 +25,7 @@ from litellm.llms.base_llm.ocr.transformation import (
 )
 from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
 from litellm.rust_bridge import ocr as rust_ocr_bridge
-from litellm.rust_bridge.dispatch import PROPAGATE, anative_first, native_first
+from litellm.rust_bridge.dispatch import anative_first, native_first, provider_errors
 from litellm.rust_bridge.runtime import DispatchResult
 from litellm.types.router import GenericLiteLLMParams
 from litellm.utils import ProviderConfigManager, client
@@ -159,7 +159,9 @@ def _prepare_ocr_request(
 @anative_first(
     native=rust_ocr_bridge.aattempt_ocr,
     route="ocr",
-    errors=lambda prepared_request, resolve_api_key: PROPAGATE,
+    errors=lambda prepared_request, resolve_api_key: provider_errors(
+        prepared_request.custom_llm_provider, prepared_request.model
+    ),
 )
 async def _execute_aocr(
     prepared_request: rust_ocr_bridge.PreparedOCRRequest,
@@ -196,7 +198,9 @@ def _attempt_ocr(
 @native_first(
     native=_attempt_ocr,
     route="ocr",
-    errors=lambda prepared_request, resolve_api_key, is_async: PROPAGATE,
+    errors=lambda prepared_request, resolve_api_key, is_async: provider_errors(
+        prepared_request.custom_llm_provider, prepared_request.model
+    ),
 )
 def _execute_ocr(
     prepared_request: rust_ocr_bridge.PreparedOCRRequest,
diff --git a/litellm/rust_bridge/chat_completions.py b/litellm/rust_bridge/chat_completions.py
index d9f06098903..c463a82c431 100644
--- a/litellm/rust_bridge/chat_completions.py
+++ b/litellm/rust_bridge/chat_completions.py
@@ -22,7 +22,6 @@ from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response impo
 from litellm.llms.bedrock.request_metadata import bedrock_request_metadata_is_owned
 from litellm.rust_bridge.bindings import UNCHANGED, NativeBinding, Unchanged
 from litellm.rust_bridge.configuration import rust_enabled
-from litellm.rust_bridge.dispatch import APIErrorMapping, ErrorAction, ErrorHandling
 from litellm.rust_bridge.protocols import (
     RustAchatCompletions,
     RustChatCompletions,
@@ -291,11 +290,3 @@ async def achat_completions(
         call=call,
         adapt=adapt,
     )
-
-
-def error_handling(provider: str, model: str) -> ErrorHandling:
-    return ErrorHandling(
-        declined=ErrorAction.SKIP,
-        upstream=APIErrorMapping(provider=provider, model=model),
-        missing_metadata=ErrorAction.SKIP,
-    )
diff --git a/litellm/rust_bridge/dispatch.py b/litellm/rust_bridge/dispatch.py
index 8c275db3533..d206eadd166 100644
--- a/litellm/rust_bridge/dispatch.py
+++ b/litellm/rust_bridge/dispatch.py
@@ -8,7 +8,7 @@ from functools import wraps
 from typing import Final, ParamSpec, TypeAlias, TypeVar
 
 from litellm._logging import verbose_logger
-from litellm.exceptions import APIError
+from litellm.exceptions import APIError, AuthenticationError, InternalServerError, RateLimitError
 from litellm.rust_bridge.bindings import native_declined_types, native_upstream_types
 from litellm.rust_bridge.runtime import DispatchResult, Handled, NativeFailed, NativeSkipped, NativeSkipReason
 
@@ -50,6 +50,13 @@ PYTHON_ON_ERROR: Final = ErrorHandling(
 )
 
 
+def provider_errors(provider: str, model: str) -> ErrorHandling:
+    return ErrorHandling(
+        declined=ErrorAction.SKIP,
+        upstream=APIErrorMapping(provider=provider, model=model),
+    )
+
+
 def _handle_error(error: Exception, action: FailureAction, route: str, reason: NativeSkipReason) -> NativeSkipped:
     match action:
         case ErrorAction.SKIP:
@@ -66,9 +73,16 @@ def _handle_error(error: Exception, action: FailureAction, route: str, reason: N
             )
             status: Final = status_value if isinstance(status_value, int) else 0
             message: Final = message_value if isinstance(message_value, str) else str(message_value)
+            error_message: Final = f"litellm rust {route}: {message}"
+            if status == 401:
+                raise AuthenticationError(message=error_message, llm_provider=provider, model=model) from error
+            if status == 429:
+                raise RateLimitError(message=error_message, llm_provider=provider, model=model) from error
+            if status == 500:
+                raise InternalServerError(message=error_message, llm_provider=provider, model=model) from error
             raise APIError(
                 status_code=status or 500,
-                message=f"litellm rust {route}: {message}",
+                message=error_message,
                 llm_provider=provider,
                 model=model,
             ) from error
diff --git a/tests/test_litellm/anthropic_interface/test_rust_bridge_messages.py b/tests/test_litellm/anthropic_interface/test_rust_bridge_messages.py
index d2096bc515b..ec198939806 100644
--- a/tests/test_litellm/anthropic_interface/test_rust_bridge_messages.py
+++ b/tests/test_litellm/anthropic_interface/test_rust_bridge_messages.py
@@ -400,17 +400,54 @@ async def test_gate_falls_back_when_bridge_unavailable(monkeypatch):
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("selection", ("native", "disabled", "failed"))
-async def test_messages_handler_runs_selected_backend_once(selection: str) -> None:
+@pytest.mark.parametrize("selection", ("native", "disabled", "failed", "declined", "upstream"))
+async def test_messages_handler_runs_selected_backend_once(selection: str, monkeypatch: pytest.MonkeyPatch) -> None:
     from datetime import datetime
+    from types import SimpleNamespace
 
     import httpx
 
+    from litellm.exceptions import RateLimitError
     from litellm.litellm_core_utils.litellm_logging import Logging
     from litellm.llms.anthropic.experimental_pass_through.messages.transformation import AnthropicMessagesConfig
     from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
+    from litellm.rust_bridge import bindings
 
-    bridge = RaisingAsyncMessages() if selection == "failed" else RecordingAsyncMessages()
+    class Declined(Exception):
+        pass
+
+    class Upstream(Exception):
+        pass
+
+    error = (
+        Upstream(429, "rate limited")
+        if selection == "upstream"
+        else Declined("unsupported")
+        if selection == "declined"
+        else RuntimeError("native failed")
+        if selection == "failed"
+        else None
+    )
+
+    class Native:
+        def __init__(self) -> None:
+            self.calls = 0
+
+        async def __call__(self, **kwargs: object) -> dict[str, object]:
+            self.calls += 1
+            if error is not None:
+                raise error
+            return dict(FAKE_MESSAGES_RESPONSE)
+
+    bridge = Native()
+    monkeypatch.setattr(
+        bindings,
+        "get_native_bridge",
+        lambda: SimpleNamespace(
+            RustBridgeDeclined=Declined,
+            RustUpstreamError=Upstream,
+        ),
+    )
     rust_messages.set_rust_messages(amessages=bridge)
     litellm.rust(selection != "disabled")
     requests: list[httpx.Request] = []
@@ -432,20 +469,32 @@ async def test_messages_handler_runs_selected_backend_once(selection: str) -> No
     await client.client.aclose()
     async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as transport:
         client.client = transport
-        response = await BaseLLMHTTPHandler().async_anthropic_messages_handler(
-            model=FAKE_MESSAGES_RESPONSE["model"],
-            messages=[{"role": "user", "content": "hello"}],
-            anthropic_messages_provider_config=AnthropicMessagesConfig(),
-            anthropic_messages_optional_request_params={"max_tokens": 10},
-            custom_llm_provider="anthropic",
-            litellm_params=GenericLiteLLMParams(),
-            logging_obj=logging_obj,
-            api_key="sk-test",
-            api_base="https://example.test",
-            client=client,
-        )
-    assert response["id"] == FAKE_MESSAGES_RESPONSE["id"]
-    assert len(requests) == (0 if selection == "native" else 1)
-    assert (bridge.calls if isinstance(bridge, RaisingAsyncMessages) else len(bridge.calls)) == (
-        0 if selection == "disabled" else 1
-    )
+
+        async def run():
+            return await BaseLLMHTTPHandler().async_anthropic_messages_handler(
+                model=FAKE_MESSAGES_RESPONSE["model"],
+                messages=[{"role": "user", "content": "hello"}],
+                anthropic_messages_provider_config=AnthropicMessagesConfig(),
+                anthropic_messages_optional_request_params={"max_tokens": 10},
+                custom_llm_provider="anthropic",
+                litellm_params=GenericLiteLLMParams(),
+                logging_obj=logging_obj,
+                api_key="sk-test",
+                api_base="https://example.test",
+                client=client,
+            )
+
+        if selection in ("failed", "upstream"):
+            with pytest.raises(RateLimitError if selection == "upstream" else RuntimeError) as caught:
+                await run()
+            if selection == "upstream":
+                assert caught.value.__cause__ is error
+                assert caught.value.llm_provider == "anthropic"
+                assert caught.value.model == FAKE_MESSAGES_RESPONSE["model"]
+            else:
+                assert caught.value is error
+        else:
+            response = await run()
+            assert response["id"] == FAKE_MESSAGES_RESPONSE["id"]
+    assert len(requests) == (1 if selection in ("disabled", "declined") else 0)
+    assert bridge.calls == (0 if selection == "disabled" else 1)
diff --git a/tests/test_litellm/responses/test_rust_bridge_websocket.py b/tests/test_litellm/responses/test_rust_bridge_websocket.py
index e502689797b..9752f95198f 100644
--- a/tests/test_litellm/responses/test_rust_bridge_websocket.py
+++ b/tests/test_litellm/responses/test_rust_bridge_websocket.py
@@ -142,3 +142,30 @@ async def test_managed_connection_closes_native_socket_on_consumer_failure() ->
         await use_connection()
     assert socket.sent == ["hello"]
     assert socket.closed
+
+
+@pytest.mark.asyncio
+async def test_connection_failure_does_not_authorize_python_fallback() -> None:
+    from contextlib import AbstractAsyncContextManager
+
+    from litellm.rust_bridge.dispatch import anative_context, provider_errors
+
+    configuration.rust(True)
+    responses_websocket.set_rust_responses_websocket(connection=_FailingNativeBridge)
+
+    @anative_context(
+        native=lambda: responses_websocket.managed_connect(
+            url="wss://example.test/responses", headers={}, timeout=None
+        ),
+        route="responses_websocket",
+        errors=lambda: provider_errors("openai", "responses websocket"),
+    )
+    def execute() -> AbstractAsyncContextManager[object]:
+        pytest.fail("unknown native failures must not open a Python connection")
+
+    async def run() -> None:
+        async with execute():
+            pytest.fail("connection must fail before entering its body")
+
+    with pytest.raises(RuntimeError, match="connection failed"):
+        await run()
diff --git a/tests/test_litellm/rust_bridge/test_dispatch.py b/tests/test_litellm/rust_bridge/test_dispatch.py
index 9258372fb93..cd44afc5399 100644
--- a/tests/test_litellm/rust_bridge/test_dispatch.py
+++ b/tests/test_litellm/rust_bridge/test_dispatch.py
@@ -7,10 +7,9 @@ from typing import Final
 
 import pytest
 
-from litellm.exceptions import APIError
+from litellm.exceptions import APIError, AuthenticationError, InternalServerError, RateLimitError
 from litellm.rust_bridge import bindings
-from litellm.rust_bridge.chat_completions import error_handling
-from litellm.rust_bridge.dispatch import PROPAGATE, PYTHON_ON_ERROR, anative_first, native_first
+from litellm.rust_bridge.dispatch import PROPAGATE, PYTHON_ON_ERROR, anative_first, native_first, provider_errors
 from litellm.rust_bridge.runtime import DispatchResult, Handled, NativeFailed, NativeSkipped, NativeSkipReason
 
 
@@ -87,7 +86,7 @@ async def test_native_success_does_not_run_python_even_when_value_is_none(asynch
 @pytest.mark.parametrize("asynchronous", (False, True))
 @pytest.mark.parametrize("policy", ("chat", "propagate", "python"))
 @pytest.mark.parametrize("kind", ("declined", "upstream", "unknown", "unexpected", "missing"))
-async def test_declarations_preserve_endpoint_error_behavior(
+async def test_declarations_control_endpoint_error_behavior(
     monkeypatch: pytest.MonkeyPatch, asynchronous: bool, policy: str, kind: str
 ) -> None:
     if kind == "missing":
@@ -100,7 +99,7 @@ async def test_declarations_preserve_endpoint_error_behavior(
         else RuntimeError("failed")
     )
     rules: Final = (
-        error_handling("anthropic", "model")
+        provider_errors("anthropic", "model")
         if policy == "chat"
         else PYTHON_ON_ERROR
         if policy == "python"
@@ -128,11 +127,11 @@ async def test_declarations_preserve_endpoint_error_behavior(
             return await anative_first(native=anative, route="chat_completions", errors=lambda: rules)(apython)()
         return native_first(native=native, route="chat_completions", errors=lambda: rules)(python)()
 
-    if policy == "python" or (policy == "chat" and kind in ("declined", "missing")):
+    if policy == "python" or (policy == "chat" and kind == "declined"):
         assert await run() == "python response"
         assert calls == ["python"]
     elif policy == "chat" and kind == "upstream":
-        with pytest.raises(APIError) as caught:
+        with pytest.raises(RateLimitError) as caught:
             await run()
         assert caught.value.status_code == 429
         assert caught.value.model == "model"
@@ -188,15 +187,43 @@ async def test_cancellation_does_not_run_python() -> None:
         await anative_first(native=native, route="test", errors=lambda: PYTHON_ON_ERROR)(python)()
 
 
-@pytest.mark.parametrize("status", (0, 401, 403, 429, 500, 503))
-def test_chat_upstream_mapping_preserves_status_message_and_context(status: int) -> None:
+@pytest.mark.asyncio
+@pytest.mark.parametrize("asynchronous", (False, True))
+@pytest.mark.parametrize(
+    "status,exception_type",
+    (
+        (0, APIError),
+        (401, AuthenticationError),
+        (403, APIError),
+        (429, RateLimitError),
+        (500, InternalServerError),
+        (503, APIError),
+    ),
+)
+async def test_upstream_mapping_preserves_status_message_and_context(
+    asynchronous: bool, status: int, exception_type: type[Exception]
+) -> None:
     error: Final = Upstream(status, "upstream failed")
-    with pytest.raises(APIError, match="upstream failed") as caught:
-        native_first(
+
+    async def native() -> DispatchResult[str]:
+        return NativeFailed(error)
+
+    async def python() -> str:
+        pytest.fail("upstream errors must not run Python")
+
+    async def run() -> str:
+        if asynchronous:
+            return await anative_first(
+                native=native, route="chat_completions", errors=lambda: provider_errors("anthropic", "model")
+            )(python)()
+        return native_first(
             native=lambda: NativeFailed(error),
             route="chat_completions",
-            errors=lambda: error_handling("anthropic", "model"),
+            errors=lambda: provider_errors("anthropic", "model"),
         )(lambda: pytest.fail("upstream errors must not run Python"))()
+
+    with pytest.raises(exception_type, match="upstream failed") as caught:
+        await run()
     assert caught.value.status_code == (status or 500)
     assert caught.value.model == "model"
     assert caught.value.llm_provider == "anthropic"
@@ -220,7 +247,7 @@ async def test_registered_wrapper_preserves_arguments_and_request_error_context(
         return native(provider, model=model)
 
     def rules(provider: str, *, model: str):
-        return error_handling(provider, model)
+        return provider_errors(provider, model)
 
     @native_first(native=native, route="chat_completions", errors=rules)
     def execute(provider: str, *, model: str) -> str:
@@ -240,7 +267,7 @@ async def test_registered_wrapper_preserves_arguments_and_request_error_context(
         else:
             execute("second", model="limited")
 
-    with pytest.raises(APIError) as caught:
+    with pytest.raises(RateLimitError) as caught:
         await fail()
     assert caught.value.llm_provider == "second"
     assert caught.value.model == "limited"