fix(inference): honor Retry-After in the streaming retry loop (#4895) (#4909)

Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>

src/openhuman/inference/provider/error_classify.rs

@@ -112,6 +112,13 @@ pub(crate) fn is_non_retryable(err: &anyhow::Error) -> bool {
 /// Classify a StreamError without losing type information.
 /// Inspects the inner reqwest::Error status directly for Http variants.
 pub(crate) fn is_stream_error_non_retryable(err: &StreamError) -> bool {
+    // A plan/quota `429` is terminal even though `429` is normally retryable —
+    // match the non-streaming loop so a plan-restricted stream fails fast with a
+    // clear message instead of burning its (now larger) rate-limit retry budget
+    // (#4895).
+    if is_stream_non_retryable_rate_limit(err) {
+        return true;
+    }
     match err {
         StreamError::Http(reqwest_err) => {
             if let Some(status) = reqwest_err.status() {
@@ -197,6 +204,13 @@ pub(crate) fn is_upstream_unhealthy(err: &anyhow::Error) -> bool {
         || lower.contains("504 gateway timeout")
 }
 
+/// Text-only rate-limit heuristic, shared by the anyhow and streaming
+/// classifiers so both read the same `429` signal.
+fn msg_is_rate_limited(msg: &str) -> bool {
+    msg.contains("429")
+        && (msg.contains("Too Many") || msg.contains("rate") || msg.contains("limit"))
+}
+
 /// Check if an error is a rate-limit (429) error.
 pub(crate) fn is_rate_limited(err: &anyhow::Error) -> bool {
     if let Some(reqwest_err) = err.downcast_ref::<reqwest::Error>() {
@@ -204,9 +218,7 @@ pub(crate) fn is_rate_limited(err: &anyhow::Error) -> bool {
             return status.as_u16() == 429;
         }
     }
-    let msg = err.to_string();
-    msg.contains("429")
-        && (msg.contains("Too Many") || msg.contains("rate") || msg.contains("limit"))
+    msg_is_rate_limited(&err.to_string())
 }
 
 /// Check if a 429 is a business/quota-plan error that retries cannot fix.
@@ -216,11 +228,15 @@ pub(crate) fn is_rate_limited(err: &anyhow::Error) -> bool {
 /// - insufficient balance / package not active
 /// - known provider business codes (e.g. Z.AI: 1311, 1113)
 pub(crate) fn is_non_retryable_rate_limit(err: &anyhow::Error) -> bool {
-    if !is_rate_limited(err) {
-        return false;
-    }
+    is_rate_limited(err) && msg_indicates_non_retryable_rate_limit(&err.to_string())
+}
 
-    let msg = err.to_string();
+/// Core business/plan/quota-`429` detector, shared by the anyhow
+/// ([`is_non_retryable_rate_limit`]) and streaming
+/// ([`is_stream_non_retryable_rate_limit`]) classifiers so a plan/quota refusal
+/// fails fast on **both** the streaming and non-streaming paths (#4895). The
+/// caller is responsible for having already confirmed the error is a `429`.
+fn msg_indicates_non_retryable_rate_limit(msg: &str) -> bool {
     let lower = msg.to_lowercase();
 
     let business_hints = [
@@ -255,39 +271,139 @@ pub(crate) fn is_non_retryable_rate_limit(err: &anyhow::Error) -> bool {
     false
 }
 
+/// Whether a [`StreamError`] is a `429` rate-limit. Reads the typed reqwest
+/// status for `Http`, and the same text signal as [`is_rate_limited`] for the
+/// `Provider` string envelope (the managed backend surfaces its `429` body —
+/// including `errorCode`/`retryAfter` — through `StreamError::Provider`).
+pub(crate) fn is_stream_rate_limited(err: &StreamError) -> bool {
+    match err {
+        StreamError::Http(reqwest_err) => reqwest_err
+            .status()
+            .is_some_and(|status| status.as_u16() == 429),
+        StreamError::Provider(msg) => msg_is_rate_limited(msg),
+        _ => false,
+    }
+}
+
+/// Whether a streaming `429` is a business/plan/quota refusal that retries
+/// cannot fix (the streaming analogue of [`is_non_retryable_rate_limit`]).
+pub(crate) fn is_stream_non_retryable_rate_limit(err: &StreamError) -> bool {
+    is_stream_rate_limited(err) && msg_indicates_non_retryable_rate_limit(&err.to_string())
+}
+
+/// Extract a `Retry-After` (milliseconds) carried by a streaming error. The
+/// managed backend embeds its rate-limit `retryAfter` (and any provider
+/// `Retry-After` header text) in the `StreamError::Provider` envelope, so parse
+/// it from the error's `Display` string.
+pub(crate) fn parse_stream_retry_after_ms(err: &StreamError) -> Option<u64> {
+    parse_retry_after_ms_from_str_at(&err.to_string(), chrono::Utc::now())
+}
+
+/// Streaming-retry backoff (ms): honor a server `Retry-After` (capped at
+/// [`RETRY_AFTER_CAP_MS`], floored at `base`) when present, else the caller's
+/// exponential `base`. Mirrors `ReliableProvider::compute_backoff` for the
+/// streaming path, which previously ignored `Retry-After` entirely (#4895).
+pub(crate) fn compute_stream_backoff_ms(base: u64, err: &StreamError) -> u64 {
+    match parse_stream_retry_after_ms(err) {
+        Some(retry_after) => retry_after.min(RETRY_AFTER_CAP_MS).max(base),
+        None => base,
+    }
+}
+
+/// Cap on any honored `Retry-After` wait, in milliseconds. Mirrors the
+/// non-streaming `ReliableProvider::compute_backoff` cap and
+/// `agent::triage::evaluator::RETRY_AFTER_CAP` so a hostile or mis-set header
+/// can't wedge a turn for minutes.
+pub(crate) const RETRY_AFTER_CAP_MS: u64 = 30_000;
+
 /// Try to extract a Retry-After value (in milliseconds) from an error message.
-/// Looks for patterns like `Retry-After: 5` or `retry_after: 2.5` in the error string.
+/// Looks for patterns like `Retry-After: 5` or `retry_after: 2.5` in the error
+/// string. Convenience wrapper over [`parse_retry_after_ms_from_str_at`] using
+/// the current wall-clock (only relevant for the HTTP-date form).
 pub(crate) fn parse_retry_after_ms(err: &anyhow::Error) -> Option<u64> {
-    let msg = err.to_string();
+    parse_retry_after_ms_from_str_at(&err.to_string(), chrono::Utc::now())
+}
+
+/// Extract a `Retry-After` value (milliseconds) from a raw error/body string.
+///
+/// Recognises, case-insensitively:
+/// - the HTTP header form `Retry-After: <secs>` / `retry_after <secs>`
+///   (integer or fractional **seconds**),
+/// - the backend JSON body field the managed proxy emits,
+///   `"retryAfter": <secs>` (camelCase, see `error_code.rs`), and
+/// - an HTTP-date (RFC 7231 IMF-fixdate, e.g. `Wed, 21 Oct 2025 07:28:00 GMT`),
+///   whose delay is computed relative to `now` and floored at zero.
+///
+/// `now` is injected so the HTTP-date branch is deterministically testable.
+pub(crate) fn parse_retry_after_ms_from_str_at(
+    msg: &str,
+    now: chrono::DateTime<chrono::Utc>,
+) -> Option<u64> {
     let lower = msg.to_lowercase();
 
-    // Look for "retry-after: <number>" or "retry_after: <number>"
-    for prefix in &[
-        "retry-after:",
-        "retry_after:",
-        "retry-after ",
-        "retry_after ",
-    ] {
-        if let Some(pos) = lower.find(prefix) {
-            let after = &msg[pos + prefix.len()..];
-            let num_str: String = after
-                .trim()
-                .chars()
-                .take_while(|c| c.is_ascii_digit() || *c == '.')
-                .collect();
+    // Base keys without a trailing separator, so a single pass covers the
+    // header (`Retry-After: 5`), space (`retry-after 5`), and JSON
+    // (`"retryAfter":30`) spellings once the punctuation is trimmed.
+    for key in &["retry-after", "retry_after", "retryafter"] {
+        let Some(pos) = lower.find(key) else {
+            continue;
+        };
+        // Work on the original-case slice so an HTTP-date still parses. Use
+        // `get` (not direct slicing) so a non-ASCII byte earlier in the string —
+        // which would make the lowercased `pos` land off a char boundary — yields
+        // no match instead of panicking.
+        let Some(after) = msg.get(pos + key.len()..) else {
+            continue;
+        };
+        let value = after
+            .trim_start_matches(|c: char| c == '"' || c == ':' || c == '=' || c.is_whitespace());
+
+        // Numeric seconds (integer or fractional) first.
+        let num_str: String = value
+            .chars()
+            .take_while(|c| c.is_ascii_digit() || *c == '.')
+            .collect();
+        if !num_str.is_empty() {
             if let Ok(secs) = num_str.parse::<f64>() {
                 if secs.is_finite() && secs >= 0.0 {
-                    let millis = Duration::from_secs_f64(secs).as_millis();
-                    if let Ok(value) = u64::try_from(millis) {
+                    if let Ok(value) = u64::try_from(Duration::from_secs_f64(secs).as_millis()) {
                         return Some(value);
                     }
                 }
             }
         }
+
+        // Otherwise try an HTTP-date value.
+        if let Some(ms) = parse_http_date_delay_ms(value, now) {
+            return Some(ms);
+        }
     }
     None
 }
 
+/// Parse an HTTP-date `Retry-After` value (RFC 7231 IMF-fixdate) and return the
+/// delay from `now` in milliseconds (floored at zero for past dates), or `None`
+/// when `value` doesn't lead with a parseable GMT date.
+fn parse_http_date_delay_ms(value: &str, now: chrono::DateTime<chrono::Utc>) -> Option<u64> {
+    let trimmed = value.trim();
+    // Bound the candidate to the date itself (ends at the "GMT" zone marker) so
+    // trailing JSON/prose (`… GMT"}`) doesn't defeat the strict parser.
+    let end = trimmed.find("GMT")? + 3;
+    // IMF-fixdate ("… 07:28:00 GMT") differs from RFC 2822 only in the zone
+    // spelling; normalise GMT → +0000 so chrono's RFC 2822 parser accepts it.
+    let normalized = trimmed[..end].replace("GMT", "+0000");
+    let parsed = chrono::DateTime::parse_from_rfc2822(normalized.trim()).ok()?;
+    let delta_ms = parsed
+        .with_timezone(&chrono::Utc)
+        .signed_duration_since(now)
+        .num_milliseconds();
+    if delta_ms <= 0 {
+        Some(0)
+    } else {
+        u64::try_from(delta_ms).ok()
+    }
+}
+
 pub(crate) fn failure_reason(
     rate_limited: bool,
     non_retryable: bool,
@@ -635,6 +751,114 @@ mod tests {
         );
     }
 
+    // ── #4895: streaming Retry-After honoring + plan-vs-transient 429 split ──
+
+    #[test]
+    fn parse_retry_after_json_body_field() {
+        let now = chrono::Utc::now();
+        // The managed backend embeds `retryAfter` (camelCase, seconds) in the
+        // JSON error body that reaches the loop as a `StreamError::Provider`.
+        let body = r#"OpenHuman API error (429 Too Many Requests): {"error":{"message":"slow down","errorCode":"RATE_LIMITED","retryAfter":30}}"#;
+        assert_eq!(
+            parse_retry_after_ms_from_str_at(body, now),
+            Some(30_000),
+            "camelCase JSON retryAfter must be honored"
+        );
+        // Pretty-printed body with a space after the colon.
+        let spaced = r#"{"errorCode":"RATE_LIMITED","retryAfter": 12}"#;
+        assert_eq!(parse_retry_after_ms_from_str_at(spaced, now), Some(12_000));
+    }
+
+    #[test]
+    fn parse_retry_after_header_forms_still_work() {
+        let now = chrono::Utc::now();
+        assert_eq!(
+            parse_retry_after_ms_from_str_at("429 Too Many Requests, Retry-After: 5", now),
+            Some(5_000)
+        );
+        assert_eq!(
+            parse_retry_after_ms_from_str_at("rate limited. retry_after: 2.5 seconds", now),
+            Some(2_500)
+        );
+        assert_eq!(
+            parse_retry_after_ms_from_str_at("500 Internal Server Error", now),
+            None
+        );
+    }
+
+    #[test]
+    fn parse_retry_after_http_date_form() {
+        // Anchor on chrono's own RFC-2822 parse so the fixture stays consistent
+        // with the parser (and the weekday can't silently drift).
+        let target = chrono::DateTime::parse_from_rfc2822("Tue, 21 Oct 2025 07:28:00 +0000")
+            .expect("valid rfc2822 anchor")
+            .with_timezone(&chrono::Utc);
+        let msg = "OpenHuman API error (429): Retry-After: Tue, 21 Oct 2025 07:28:00 GMT";
+
+        // 5s before the target → ~5000ms wait.
+        let now = target - chrono::Duration::seconds(5);
+        assert_eq!(parse_retry_after_ms_from_str_at(msg, now), Some(5_000));
+
+        // A date already in the past floors to 0 (never negative).
+        let past_now = target + chrono::Duration::seconds(10);
+        assert_eq!(parse_retry_after_ms_from_str_at(msg, past_now), Some(0));
+    }
+
+    #[test]
+    fn stream_rate_limited_detection() {
+        assert!(is_stream_rate_limited(&StreamError::Provider(
+            "OpenHuman API error (429 Too Many Requests): rate limit".into()
+        )));
+        assert!(!is_stream_rate_limited(&StreamError::Provider(
+            "500 Internal Server Error".into()
+        )));
+        assert!(!is_stream_rate_limited(&StreamError::InvalidSse(
+            "bad frame".into()
+        )));
+    }
+
+    #[test]
+    fn stream_non_retryable_rate_limit_splits_plan_from_transient() {
+        // Plan/quota 429 → terminal (retries can't fix it).
+        let plan = StreamError::Provider(
+            r#"OpenHuman API error (429 Too Many Requests): {"code":1311,"message":"the current account plan does not include glm-5"}"#
+                .into(),
+        );
+        assert!(is_stream_non_retryable_rate_limit(&plan));
+        assert!(
+            is_stream_error_non_retryable(&plan),
+            "a plan 429 must fail fast on the streaming path too"
+        );
+
+        // Transient 429 → retryable (must NOT be flagged terminal).
+        let transient =
+            StreamError::Provider("OpenHuman API error (429 Too Many Requests): slow down".into());
+        assert!(!is_stream_non_retryable_rate_limit(&transient));
+        assert!(
+            !is_stream_error_non_retryable(&transient),
+            "a transient 429 must remain retryable so it can be backed off"
+        );
+    }
+
+    #[test]
+    fn compute_stream_backoff_honors_and_caps_retry_after() {
+        let base = 500;
+        // Retry-After present → honored (floored at base).
+        let five_s = StreamError::Provider(
+            r#"OpenHuman API error (429): {"errorCode":"RATE_LIMITED","retryAfter":5}"#.into(),
+        );
+        assert_eq!(compute_stream_backoff_ms(base, &five_s), 5_000);
+        // Oversized Retry-After is capped.
+        let huge = StreamError::Provider(r#"{"errorCode":"RATE_LIMITED","retryAfter":600}"#.into());
+        assert_eq!(compute_stream_backoff_ms(base, &huge), RETRY_AFTER_CAP_MS);
+        // Tiny Retry-After is floored at the caller's base backoff.
+        let tiny = StreamError::Provider(r#"{"retryAfter":0}"#.into());
+        assert_eq!(compute_stream_backoff_ms(base, &tiny), base);
+        // No Retry-After → base backoff unchanged.
+        let none = StreamError::Provider("transient upstream blip".into());
+        assert_eq!(compute_stream_backoff_ms(base, &none), base);
+    }
+
     // ── upstream_unhealthy classification and failure_reason precedence ──
 
     #[test]

src/openhuman/inference/provider/reliable.rs

@@ -39,6 +39,14 @@ use std::time::Duration;
 // `reliable::format_failure_aggregate` import paths continue to resolve.
 pub(crate) use super::error_classify::*;
 
+/// Minimum retry budget for a **transient** streaming `429`, so a multi-second
+/// server `Retry-After` window can actually be waited out. The configured
+/// `provider_retries` (default 2, ~1.5 s of fixed backoff) is too small to ride
+/// out a rate-limit window; rate-limited streams get at least this many retries
+/// while every other failure keeps the configured budget (#4895). Bounded, and
+/// only applies to retryable (non-plan/quota) `429`s.
+const STREAM_RATE_LIMIT_MIN_RETRIES: u32 = 3;
+
 fn push_failure(
     failures: &mut Vec<String>,
     provider_name: &str,
@@ -825,6 +833,10 @@ impl Provider for ReliableProvider {
         let max_retries = self.max_retries;
 
         tokio::spawn(async move {
+            // Tracks whether the *last* candidate error was a rate-limit (429),
+            // so the terminal message surfaced on exhaustion is a clear
+            // rate-limit notice rather than a generic failure (#4895).
+            let mut last_error_rate_limited = false;
             for (provider_name, provider, current_model) in candidates {
                 let mut backoff_ms = base_backoff_ms;
                 let mut attempts = 0u32;
@@ -859,6 +871,19 @@ impl Provider for ReliableProvider {
                         }
                         Some(Err(ref e)) => {
                             let non_retryable = is_stream_error_non_retryable(e);
+                            // A retryable (transient) 429 — as opposed to a
+                            // plan/quota 429, which `is_stream_error_non_retryable`
+                            // already flags — gets a larger, bounded retry budget
+                            // so a multi-second Retry-After window can be waited
+                            // out (#4895).
+                            let retryable_rate_limited =
+                                !non_retryable && is_stream_rate_limited(e);
+                            last_error_rate_limited = is_stream_rate_limited(e);
+                            let effective_max = if retryable_rate_limited {
+                                max_retries.max(STREAM_RATE_LIMIT_MIN_RETRIES)
+                            } else {
+                                max_retries
+                            };
 
                             tracing::warn!(
                                 provider = provider_name,
@@ -868,12 +893,17 @@ impl Provider for ReliableProvider {
                                 "Streaming failed{}", if non_retryable { " (non-retryable)" } else { "" }
                             );
 
-                            if non_retryable || attempts >= max_retries {
+                            if non_retryable || attempts >= effective_max {
                                 break; // Move to next candidate
                             }
 
                             attempts += 1;
-                            tokio::time::sleep(Duration::from_millis(backoff_ms)).await;
+                            // Honor the server's Retry-After (capped) on
+                            // rate-limits instead of the raw exponential doubling,
+                            // which previously fired all attempts within ~1.5 s and
+                            // ignored the backend's requested wait entirely (#4895).
+                            let wait = compute_stream_backoff_ms(backoff_ms, e);
+                            tokio::time::sleep(Duration::from_millis(wait)).await;
                             backoff_ms = (backoff_ms.saturating_mul(2)).min(10_000);
                             // Re-create the candidate stream on the next iteration.
                             continue;
@@ -893,11 +923,20 @@ impl Provider for ReliableProvider {
                 }
             }
 
-            // All providers/models exhausted
+            // All providers/models exhausted. When the terminal failure was a
+            // rate-limit, surface a clear, user-actionable message instead of a
+            // generic one so the turn no longer dies silently (#4895) — this
+            // string propagates to the chat error surface via
+            // `crate_provider`'s `ProviderFailed` → `bail!` mapping.
+            let terminal = if last_error_rate_limited {
+                "You're being rate-limited (sending requests faster than your current plan allows). \
+                 Please wait a few seconds and try again."
+                    .to_string()
+            } else {
+                "All streaming providers/models failed".to_string()
+            };
             let _ = tx
-                .send(Err(super::traits::StreamError::Provider(
-                    "All streaming providers/models failed".to_string(),
-                )))
+                .send(Err(super::traits::StreamError::Provider(terminal)))
                 .await;
         });
 

src/openhuman/inference/provider/reliable_tests.rs

@@ -1029,3 +1029,200 @@ async fn chat_with_system_bail_omits_hint_when_fallbacks_configured_but_all_fail
         "expected dump to mention every model tried: {err}"
     );
 }
+
+// ── #4895: streaming rate-limit (429) Retry-After honoring + budget/message ──
+
+/// Streaming mock that fails with a configurable `StreamError::Provider` for the
+/// first `fail_until` stream creations, then yields a single `"hello"` chunk.
+/// Mirrors the real provider's one-item-per-stream shape so a retry that
+/// re-polled a dead stream would see `None` and give up.
+struct StreamingRateLimitMock {
+    stream_calls: Arc<AtomicUsize>,
+    fail_until: usize,
+    error: String,
+}
+
+#[async_trait]
+impl Provider for StreamingRateLimitMock {
+    async fn chat_with_system(
+        &self,
+        _system_prompt: Option<&str>,
+        _message: &str,
+        _model: &str,
+        _temperature: f64,
+    ) -> anyhow::Result<String> {
+        anyhow::bail!("unused")
+    }
+
+    async fn chat_with_history(
+        &self,
+        _messages: &[ChatMessage],
+        _model: &str,
+        _temperature: f64,
+    ) -> anyhow::Result<String> {
+        anyhow::bail!("unused")
+    }
+
+    fn supports_streaming(&self) -> bool {
+        true
+    }
+
+    fn stream_chat_with_system(
+        &self,
+        _system_prompt: Option<&str>,
+        _message: &str,
+        _model: &str,
+        _temperature: f64,
+        _options: StreamOptions,
+    ) -> futures_util::stream::BoxStream<'static, StreamResult<StreamChunk>> {
+        use futures_util::{stream, StreamExt};
+        let n = self.stream_calls.fetch_add(1, Ordering::SeqCst) + 1;
+        let succeed = n > self.fail_until;
+        let error = self.error.clone();
+        stream::once(async move {
+            if succeed {
+                Ok(StreamChunk::delta("hello"))
+            } else {
+                Err(StreamError::Provider(error))
+            }
+        })
+        .boxed()
+    }
+}
+
+/// A transient streaming `429` carrying a `Retry-After` must be retried, must
+/// actually WAIT the server's requested window (not the old ~1.5 s fixed
+/// backoff), and then recover. Uses tokio's paused clock so the 5 s wait is
+/// virtual/instant yet asserted via virtual elapsed time.
+#[tokio::test(start_paused = true)]
+async fn streaming_rate_limit_honors_retry_after_and_recovers() {
+    use futures_util::StreamExt;
+
+    let stream_calls = Arc::new(AtomicUsize::new(0));
+    let provider = ReliableProvider::new(
+        vec![(
+            "openhuman".into(),
+            Box::new(StreamingRateLimitMock {
+                stream_calls: Arc::clone(&stream_calls),
+                fail_until: 1,
+                error: r#"OpenHuman API error (429 Too Many Requests): {"errorCode":"RATE_LIMITED","retryAfter":5}"#
+                    .to_string(),
+            }),
+        )],
+        2,
+        50,
+    );
+
+    let start = tokio::time::Instant::now();
+    let mut stream =
+        provider.stream_chat_with_system(None, "hi", "reasoning-v1", 0.0, StreamOptions::new(true));
+    let mut chunks = Vec::new();
+    while let Some(item) = stream.next().await {
+        if let Ok(chunk) = item {
+            chunks.push(chunk.delta);
+        }
+    }
+    let elapsed = start.elapsed();
+
+    assert_eq!(
+        chunks,
+        vec!["hello".to_string()],
+        "a transient 429 with Retry-After must be retried and recover"
+    );
+    assert_eq!(
+        stream_calls.load(Ordering::SeqCst),
+        2,
+        "one rate-limited attempt + one successful retry"
+    );
+    assert!(
+        elapsed >= Duration::from_secs(5),
+        "must wait the server's 5s Retry-After before retrying, waited {elapsed:?}"
+    );
+}
+
+/// A plan/quota `429` (retries can't fix it) must fail fast — a single stream
+/// creation, no retries — and surface a clear rate-limit message.
+#[tokio::test]
+async fn streaming_plan_rate_limit_fails_fast_with_clear_message() {
+    use futures_util::StreamExt;
+
+    let stream_calls = Arc::new(AtomicUsize::new(0));
+    let provider = ReliableProvider::new(
+        vec![(
+            "openhuman".into(),
+            Box::new(StreamingRateLimitMock {
+                stream_calls: Arc::clone(&stream_calls),
+                fail_until: 100, // always fail
+                error: r#"OpenHuman API error (429 Too Many Requests): {"code":1311,"message":"the current account plan does not include glm-5"}"#
+                    .to_string(),
+            }),
+        )],
+        3,
+        50,
+    );
+
+    let mut stream =
+        provider.stream_chat_with_system(None, "hi", "reasoning-v1", 0.0, StreamOptions::new(true));
+    let mut terminal: Option<String> = None;
+    while let Some(item) = stream.next().await {
+        if let Err(StreamError::Provider(msg)) = item {
+            terminal = Some(msg);
+        }
+    }
+
+    assert_eq!(
+        stream_calls.load(Ordering::SeqCst),
+        1,
+        "a plan/quota 429 must fail fast without burning the retry budget"
+    );
+    let terminal = terminal.expect("stream must surface a terminal error");
+    assert!(
+        terminal.to_lowercase().contains("rate-limited"),
+        "plan 429 terminal must be a clear rate-limit message, got: {terminal}"
+    );
+}
+
+/// A persistently transient `429` must use the dedicated rate-limit retry budget
+/// (strictly more than the small configured `provider_retries`) before giving
+/// up with a clear terminal rate-limit message. `retryAfter:0` keeps the waits
+/// at the base backoff so the test stays fast.
+#[tokio::test(start_paused = true)]
+async fn streaming_transient_rate_limit_uses_dedicated_budget_then_clear_message() {
+    use futures_util::StreamExt;
+
+    let stream_calls = Arc::new(AtomicUsize::new(0));
+    let provider = ReliableProvider::new(
+        vec![(
+            "openhuman".into(),
+            Box::new(StreamingRateLimitMock {
+                stream_calls: Arc::clone(&stream_calls),
+                fail_until: 100, // always fail
+                error: r#"OpenHuman API error (429 Too Many Requests): {"errorCode":"RATE_LIMITED","retryAfter":0}"#
+                    .to_string(),
+            }),
+        )],
+        2, // configured retries; rate-limit floor is STREAM_RATE_LIMIT_MIN_RETRIES (3)
+        50,
+    );
+
+    let mut stream =
+        provider.stream_chat_with_system(None, "hi", "reasoning-v1", 0.0, StreamOptions::new(true));
+    let mut terminal: Option<String> = None;
+    while let Some(item) = stream.next().await {
+        if let Err(StreamError::Provider(msg)) = item {
+            terminal = Some(msg);
+        }
+    }
+
+    // 1 initial attempt + 3 rate-limit retries (dedicated budget > configured 2).
+    assert_eq!(
+        stream_calls.load(Ordering::SeqCst),
+        4,
+        "a transient 429 must use the dedicated rate-limit retry budget, not the smaller configured one"
+    );
+    let terminal = terminal.expect("stream must surface a terminal error");
+    assert!(
+        terminal.to_lowercase().contains("rate-limited"),
+        "an exhausted transient 429 must end with a clear rate-limit message, got: {terminal}"
+    );
+}