From aa54d0dcda988a9a3f2b2045672bf2e80131edd3 Mon Sep 17 00:00:00 2001 From: Mega Mind <146339422+M3gA-Mind@users.noreply.github.com> Date: Thu, 16 Jul 2026 05:35:05 +0530 Subject: [PATCH] fix(inference): honor Retry-After in the streaming retry loop (#4895) (#4909) Co-authored-by: Steven Enamakel --- .../inference/provider/error_classify.rs | 274 ++++++++++++++++-- src/openhuman/inference/provider/reliable.rs | 51 +++- .../inference/provider/reliable_tests.rs | 197 +++++++++++++ 3 files changed, 491 insertions(+), 31 deletions(-) diff --git a/src/openhuman/inference/provider/error_classify.rs b/src/openhuman/inference/provider/error_classify.rs index bd6efce23..2ebf9d32d 100644 --- a/src/openhuman/inference/provider/error_classify.rs +++ b/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::() { @@ -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 { + 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 { - 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: ` / `retry_after ` +/// (integer or fractional **seconds**), +/// - the backend JSON body field the managed proxy emits, +/// `"retryAfter": ` (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, +) -> Option { let lower = msg.to_lowercase(); - // Look for "retry-after: " or "retry_after: " - 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::() { 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) -> Option { + 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] diff --git a/src/openhuman/inference/provider/reliable.rs b/src/openhuman/inference/provider/reliable.rs index 526bc41a1..27007f6ff 100644 --- a/src/openhuman/inference/provider/reliable.rs +++ b/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, 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; }); diff --git a/src/openhuman/inference/provider/reliable_tests.rs b/src/openhuman/inference/provider/reliable_tests.rs index 377bb08e4..1692e9a78 100644 --- a/src/openhuman/inference/provider/reliable_tests.rs +++ b/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, + 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 { + anyhow::bail!("unused") + } + + async fn chat_with_history( + &self, + _messages: &[ChatMessage], + _model: &str, + _temperature: f64, + ) -> anyhow::Result { + 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> { + 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 = 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 = 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}" + ); +}