fix(llm, openai): Stop a redeemed usage reset locking the account out

crates/jp_llm/src/error.rs

@@ -39,6 +39,16 @@ pub struct StreamError {
     /// account-wide.
     pub quota_scope: Option<String>,
 
+    /// How many of the exhausted limit's usage windows the provider reported as
+    /// spent.
+    ///
+    /// A subscription limit reports a short window and a long one.
+    /// A reset credit reopens one of them, so a request stays refused while
+    /// more than one is spent.
+    ///
+    /// `0` when the provider reported no window state at all.
+    pub quota_spent_windows: usize,
+
     /// Human-readable error message.
     message: String,
 
@@ -59,6 +69,7 @@ impl StreamError {
             retry_after: None,
             quota_reset: None,
             quota_scope: None,
+            quota_spent_windows: 0,
             source: None,
         }
     }

crates/jp_llm/src/provider/openai.rs

@@ -29,6 +29,7 @@ use openai_responses::{
 use reqwest::header::{self, HeaderMap, HeaderName, HeaderValue};
 use serde::Deserialize;
 use serde_json::{Map, Value};
+use tokio::time::{Instant, sleep};
 use tracing::{debug, trace, warn};
 
 use super::{EventStream, ModelDetails, Provider};
@@ -62,6 +63,10 @@ mod fold_tests;
 #[path = "openai/latency_tests.rs"]
 mod latency_tests;
 
+#[cfg(test)]
+#[path = "openai/redeem_tests.rs"]
+mod redeem_tests;
+
 #[cfg(test)]
 #[path = "openai/switchable_tests.rs"]
 mod switchable_tests;
@@ -104,6 +109,25 @@ const PERSISTED_REASONING: &str = "persisted_reasoning";
 /// when the flag is present.
 const EXPLICIT_PROMPT_CACHING: &str = "explicit_prompt_caching";
 
+/// How long to keep retrying after a redeemed usage reset before giving up on
+/// it.
+///
+/// A redemption the account confirmed does not reach the responses host
+/// instantly, so the request that prompted it is refused again for a short
+/// while afterwards.
+/// The credit is already spent by then, and the turn is unfinished: waiting it
+/// out is what the user paid for, where abandoning the turn wastes both.
+///
+/// Bounded because the alternative explanation — a window the credit does not
+/// cover — looks identical from here, and that one never resolves.
+const RESET_PROPAGATION_BUDGET: Duration = Duration::from_mins(1);
+
+/// How long to wait between attempts while a redeemed reset propagates.
+///
+/// Fixed rather than backed off: the wait is already short, and every attempt
+/// is an admission-stage rejection that bills nothing.
+const RESET_RETRY_INTERVAL: Duration = Duration::from_secs(5);
+
 /// How often to inject a synthetic keep-alive while a tool call is streaming.
 ///
 /// OpenAI emits the `function_call_arguments` deltas for a large tool call as a
@@ -136,6 +160,15 @@ pub struct Openai {
     /// Where subscription requests go, after the environment override.
     codex_base_url: String,
 
+    /// How long a redeemed usage reset is given to reach the responses host.
+    ///
+    /// A field rather than a constant so a test can shrink the wait it is
+    /// pinning; production always takes [`RESET_PROPAGATION_BUDGET`].
+    reset_budget: Duration,
+
+    /// How long to wait between attempts while that reset propagates.
+    reset_interval: Duration,
+
     /// The clients built for the most recently resolved credential and session.
     ///
     /// A turn issues several requests around tool execution, and resolution
@@ -181,6 +214,8 @@ impl Openai {
             fixed_credential: None,
             base_url: env_override(&config.base_url_env, &config.base_url),
             codex_base_url: env_override(&config.codex_base_url_env, &config.codex_base_url),
+            reset_budget: RESET_PROPAGATION_BUDGET,
+            reset_interval: RESET_RETRY_INTERVAL,
             client_cache: Arc::new(Mutex::new(None)),
         };
 
@@ -202,10 +237,21 @@ impl Openai {
             fixed_credential: Some((credential, resolve::Attribution::default())),
             base_url: config.base_url.clone(),
             codex_base_url: config.codex_base_url.clone(),
+            reset_budget: RESET_PROPAGATION_BUDGET,
+            reset_interval: RESET_RETRY_INTERVAL,
             client_cache: Arc::new(Mutex::new(None)),
         }
     }
 
+    /// Shorten the wait for a redeemed reset, so a test pinning it runs in
+    /// milliseconds rather than the minute production allows.
+    #[cfg(test)]
+    pub(crate) fn with_reset_timing(mut self, budget: Duration, interval: Duration) -> Self {
+        self.reset_budget = budget;
+        self.reset_interval = interval;
+        self
+    }
+
     /// Build a provider around an explicit subscription credential.
     ///
     /// Test seam for recording and replaying the subscription endpoint without
@@ -245,11 +291,16 @@ impl Openai {
     ///
     /// Returns `None` when there is no chain to advance or nothing further in
     /// it, which the caller surfaces as the original, now-terminal error.
+    ///
+    /// `after_redemption` reports whether this turn already spent a reset
+    /// credit on the attempt's profile, which is what the recorded cooldown
+    /// depends on.
     async fn advance(
         &self,
         attempt: &resolve::Attempt,
         error: &StreamError,
         model: &str,
+        after_redemption: bool,
     ) -> Option<resolve::Attempt> {
         let spent = attempt.selected.as_ref()?;
 
@@ -260,6 +311,7 @@ impl Openai {
             error,
             model,
             Utc::now(),
+            after_redemption,
         )
         .await
     }
@@ -512,7 +564,10 @@ impl Provider for Openai {
             // One redemption per turn. A plan holds few credits, and a window
             // that closes again right after being reopened is not a window a
             // second credit would fix.
-            let mut redeemed = false;
+            //
+            // The deadline is how long the reopened window is given to reach
+            // the responses host before the turn stops waiting on it.
+            let mut reopened_until: Option<Instant> = None;
 
             // A credential refused or spent at admission is not a failure of
             // the request: the next entry in the chain can serve it. Each pass
@@ -577,21 +632,45 @@ impl Provider for Openai {
                         // subscription the user already paid for, instead of
                         // falling through to per-token billing with allowance
                         // still on the account.
+                        // One credit reopens one window, so a limit reporting
+                        // more than one spent window stays closed after the
+                        // redemption. Spending a credit there buys nothing and
+                        // the account has few to spend.
                         if subscription
                             && error.kind == StreamErrorKind::SubscriptionExhausted
-                            && !redeemed
+                            && reopened_until.is_none()
+                            && error.quota_spent_windows <= 1
                             && let Some(notice) =
                                 this.redeem_reset_credit(&attempt, &session).await
                         {
-                            redeemed = true;
+                            reopened_until = Some(Instant::now() + this.reset_budget);
                             for notice in notices {
                                 yield Ok(Event::Notice(notice));
                             }
                             yield Ok(Event::Notice(notice));
                             continue;
                         }
 
-                        if let Some(mut next) = this.advance(&attempt, &error, &name).await {
+                        // The window this turn reopened has not reached the
+                        // host yet. The turn is unfinished and the credit is
+                        // already spent, so it waits rather than falling
+                        // through to per-token billing or abandoning the turn
+                        // outright.
+                        if subscription
+                            && error.kind == StreamErrorKind::SubscriptionExhausted
+                            && reopened_until.is_some_and(|until| Instant::now() < until)
+                        {
+                            for notice in notices {
+                                yield Ok(Event::Notice(notice));
+                            }
+                            sleep(this.reset_interval).await;
+                            continue;
+                        }
+
+                        if let Some(mut next) = this
+                            .advance(&attempt, &error, &name, reopened_until.is_some())
+                            .await
+                        {
                             next.notices.splice(..0, notices);
                             attempt = next;
                         } else {

crates/jp_llm/src/provider/openai/rate_limits.rs

@@ -96,10 +96,15 @@ impl Snapshot {
     /// The spent window, if either is spent.
     #[must_use]
     pub fn spent(&self) -> Option<&Window> {
+        self.spent_windows().next()
+    }
+
+    /// Every spent window of this family, in report order.
+    pub fn spent_windows(&self) -> impl Iterator<Item = &Window> {
         [self.primary.as_ref(), self.secondary.as_ref()]
             .into_iter()
             .flatten()
-            .find(|window| window.is_spent())
+            .filter(|window| window.is_spent())
     }
 
     /// The fullest window worth warning about, if any.
@@ -135,6 +140,7 @@ pub fn apply(error: &mut StreamError, headers: &HeaderMap) {
     error.kind = StreamErrorKind::SubscriptionExhausted;
     error.quota_scope = Some(snapshot.scope());
     error.quota_reset = window.resets_at;
+    error.quota_spent_windows = snapshot.spent_windows().count();
 }
 
 /// The account-wide cooldown scope.

crates/jp_llm/src/provider/openai/rate_limits_tests.rs

@@ -1,6 +1,7 @@
 use reqwest::header::{HeaderName, HeaderValue};
 
 use super::*;
+use crate::StreamErrorKind;
 
 /// The header set a real `200` from the subscription endpoint carries.
 ///
@@ -50,6 +51,59 @@ fn headers(pairs: &[(&str, &str)]) -> HeaderMap {
     map
 }
 
+/// A reset credit reopens one window, so the count is what tells the caller
+/// whether spending one can unblock the request at all.
+#[test]
+fn test_apply_counts_every_spent_window_of_the_family_it_records() {
+    let mut error = StreamError::other("refused");
+
+    apply(
+        &mut error,
+        &headers(&[
+            ("x-codex-primary-used-percent", "100"),
+            ("x-codex-primary-window-minutes", "300"),
+            ("x-codex-primary-reset-after-seconds", "1800"),
+            ("x-codex-secondary-used-percent", "100"),
+            ("x-codex-secondary-window-minutes", "10080"),
+            ("x-codex-secondary-reset-after-seconds", "604800"),
+        ]),
+    );
+
+    assert_eq!(error.kind, StreamErrorKind::SubscriptionExhausted);
+    assert_eq!(error.quota_spent_windows, 2);
+}
+
+#[test]
+fn test_apply_counts_a_single_spent_window() {
+    let mut error = StreamError::other("refused");
+
+    apply(
+        &mut error,
+        &headers(&[
+            ("x-codex-primary-used-percent", "100"),
+            ("x-codex-primary-window-minutes", "10080"),
+            ("x-codex-primary-reset-after-seconds", "604800"),
+            ("x-codex-secondary-used-percent", "0"),
+            ("x-codex-secondary-window-minutes", "0"),
+            ("x-codex-secondary-reset-after-seconds", "0"),
+        ]),
+    );
+
+    assert_eq!(error.kind, StreamErrorKind::SubscriptionExhausted);
+    assert_eq!(error.quota_spent_windows, 1);
+}
+
+/// Headers reporting nothing spent leave the error alone, count included.
+#[test]
+fn test_apply_counts_nothing_when_no_window_is_spent() {
+    let mut error = StreamError::other("refused");
+
+    apply(&mut error, &headers(LIVE_HEADERS));
+
+    assert_eq!(error.kind, StreamErrorKind::Other);
+    assert_eq!(error.quota_spent_windows, 0);
+}
+
 #[test]
 fn test_parse_all_reads_both_families_from_a_live_response() {
     let snapshots = parse_all(&headers(LIVE_HEADERS));