From 2b75f0df78e342f269be44a19cd7766e7c00f880 Mon Sep 17 00:00:00 2001 From: oxoxDev <164490987+oxoxDev@users.noreply.github.com> Date: Wed, 24 Jun 2026 22:37:55 +0530 Subject: [PATCH] fix(web): reclassify budget-caused empty response as out-of-credits (#3386) (#4032) --- src/openhuman/channels/providers/web/ops.rs | 261 +++++++++++++++++- .../channels/providers/web/run_task.rs | 64 ++++- 2 files changed, 310 insertions(+), 15 deletions(-) diff --git a/src/openhuman/channels/providers/web/ops.rs b/src/openhuman/channels/providers/web/ops.rs index 88e980e5a..69a098bf0 100644 --- a/src/openhuman/channels/providers/web/ops.rs +++ b/src/openhuman/channels/providers/web/ops.rs @@ -1,5 +1,5 @@ use std::collections::HashMap; -use std::time::Duration; +use std::time::{Duration, Instant}; use once_cell::sync::Lazy; use serde_json::{json, Value}; @@ -22,6 +22,148 @@ use super::web_errors::classify_inference_error; pub(crate) static THREAD_SESSIONS: Lazy>> = Lazy::new(|| Mutex::new(HashMap::new())); +/// A recorded budget-exhausted signal: when it happened, and which provider +/// binding it happened on. The binding scopes the signal so a managed +/// out-of-credits error never mislabels a later empty turn the user has +/// re-routed to a different provider (local / BYO), whose balance is unrelated. +#[derive(Debug, Clone)] +struct BudgetSignal { + provider_binding: String, + at: Instant, +} + +/// Per-thread "recent budget-exhausted" signal (issue #3386). +/// +/// Set when a turn terminates with an inference budget-exhausted error; read by +/// a *later* turn on the same thread whose provider returned an empty 200. The +/// managed route closes the SSE cleanly under credit exhaustion (the response +/// already flushed HTTP 200, so there is no error frame and no inline budget +/// marker — `OpenHumanBilling` carries only `charged_amount_usd`). Without this +/// correlator such a budget-caused empty turn surfaces as the generic "empty +/// response" copy instead of the actionable out-of-credits copy. +/// +/// The signal is scoped to the provider binding it was recorded on: budget is a +/// per-provider fact, so a managed-route exhaustion must not reclassify an empty +/// turn the thread has since re-routed to a local / BYO provider. +/// +/// Kept in a sibling map rather than on `SessionEntry` so the signal survives +/// the de-poison session drop (an empty turn is not poisoned, but cold-boot +/// reseeds would otherwise be the wrong lifetime to hang this on). +static THREAD_BUDGET_SIGNALS: Lazy>> = + Lazy::new(|| Mutex::new(HashMap::new())); + +/// How long a recorded budget-exhausted signal stays eligible to reclassify a +/// later empty turn on the same thread. Five minutes: long enough to bridge a +/// user retry after the first out-of-credits turn, short enough that a genuine +/// empty response well after the fact isn't mislabeled. A successful turn clears +/// the signal regardless (the balance is evidently usable again). See #3386. +const BUDGET_SIGNAL_TTL: Duration = Duration::from_secs(5 * 60); + +/// What the budget-correlator should do with a terminated turn (#3386). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum BudgetCorrelation { + /// The terminal error is itself an inference budget-exhausted error: + /// record the signal and surface the budget copy. + BudgetExhausted, + /// An empty provider response coincided with a fresh same-thread budget + /// signal: surface the budget copy in place of the "empty response" copy. + UpgradeEmptyToBudget, + /// No budget correlation — pass the error through unchanged. + PassThrough, +} + +/// Pure decision for the budget-correlator, split out so the branch matrix is +/// unit-testable without a clock or the full `run_chat_task` frame. The async +/// helpers below supply `has_fresh_signal`. +pub(super) fn classify_budget_correlation( + is_budget_error: bool, + is_empty_response: bool, + has_fresh_signal: bool, +) -> BudgetCorrelation { + if is_budget_error { + BudgetCorrelation::BudgetExhausted + } else if is_empty_response && has_fresh_signal { + BudgetCorrelation::UpgradeEmptyToBudget + } else { + BudgetCorrelation::PassThrough + } +} + +/// Pure freshness predicate (age vs TTL), split out for clock-free testing. +fn budget_signal_is_fresh(age: Duration, ttl: Duration) -> bool { + age <= ttl +} + +/// Drop every expired entry from the map, not just the one being queried. +/// Without this, a thread that hits budget exhaustion and then never retries or +/// succeeds would leak its entry for the process lifetime. Called on the write +/// path so each new budget event sweeps the map. +fn prune_stale_budget_signals(signals: &mut HashMap) { + signals.retain(|_, sig| budget_signal_is_fresh(sig.at.elapsed(), BUDGET_SIGNAL_TTL)); +} + +/// Record that this thread just hit an inference budget-exhausted error on the +/// given provider binding. +pub(super) async fn record_budget_signal(thread_id: &str, provider_binding: &str) { + let mut signals = THREAD_BUDGET_SIGNALS.lock().await; + prune_stale_budget_signals(&mut signals); + signals.insert( + key_for(thread_id), + BudgetSignal { + provider_binding: provider_binding.to_string(), + at: Instant::now(), + }, + ); +} + +/// Clear any recorded budget signal for this thread — called on a successful +/// turn, where the balance is evidently usable again. +pub(super) async fn clear_budget_signal(thread_id: &str) { + let mut signals = THREAD_BUDGET_SIGNALS.lock().await; + signals.remove(&key_for(thread_id)); +} + +/// Whether this thread has a budget signal recorded within `BUDGET_SIGNAL_TTL` +/// **on the same provider binding** as the current turn. A binding mismatch or +/// an expired entry evicts it and reads as not-fresh, so a re-routed turn never +/// inherits the prior provider's exhaustion. +pub(super) async fn has_fresh_budget_signal(thread_id: &str, provider_binding: &str) -> bool { + let mut signals = THREAD_BUDGET_SIGNALS.lock().await; + let key = key_for(thread_id); + match signals.get(&key) { + Some(sig) + if sig.provider_binding == provider_binding + && budget_signal_is_fresh(sig.at.elapsed(), BUDGET_SIGNAL_TTL) => + { + true + } + Some(_) => { + signals.remove(&key); + false + } + None => false, + } +} + +/// Test-only seeder: record a budget signal on `provider_binding` aged `age` +/// into the past so expiry can be exercised without sleeping. +#[cfg(test)] +pub(super) async fn record_budget_signal_aged( + thread_id: &str, + provider_binding: &str, + age: Duration, +) { + let mut signals = THREAD_BUDGET_SIGNALS.lock().await; + let when = Instant::now().checked_sub(age).unwrap_or_else(Instant::now); + signals.insert( + key_for(thread_id), + BudgetSignal { + provider_binding: provider_binding.to_string(), + at: when, + }, + ); +} + pub(super) static IN_FLIGHT: Lazy>> = Lazy::new(|| Mutex::new(HashMap::new())); @@ -951,3 +1093,120 @@ pub async fn channel_web_cancel( "web channel cancellation processed", )) } + +#[cfg(test)] +mod budget_correlation_tests { + use super::*; + + #[test] + fn classify_budget_correlation_matrix() { + // A budget error always records + surfaces budget copy, regardless of + // the other flags. + assert_eq!( + classify_budget_correlation(true, false, false), + BudgetCorrelation::BudgetExhausted + ); + assert_eq!( + classify_budget_correlation(true, true, true), + BudgetCorrelation::BudgetExhausted + ); + // Empty response only upgrades when a fresh signal is present. + assert_eq!( + classify_budget_correlation(false, true, true), + BudgetCorrelation::UpgradeEmptyToBudget + ); + assert_eq!( + classify_budget_correlation(false, true, false), + BudgetCorrelation::PassThrough + ); + // A fresh signal without an empty response does not invent an upgrade. + assert_eq!( + classify_budget_correlation(false, false, true), + BudgetCorrelation::PassThrough + ); + // Neither flag: untouched. + assert_eq!( + classify_budget_correlation(false, false, false), + BudgetCorrelation::PassThrough + ); + } + + #[test] + fn budget_signal_is_fresh_boundary() { + let ttl = Duration::from_secs(300); + assert!(budget_signal_is_fresh(Duration::from_secs(0), ttl)); + assert!(budget_signal_is_fresh(Duration::from_secs(299), ttl)); + assert!(budget_signal_is_fresh(ttl, ttl)); // inclusive at the boundary + assert!(!budget_signal_is_fresh(Duration::from_secs(301), ttl)); + } + + const BINDING: &str = "openhuman-managed"; + + #[tokio::test] + async fn record_then_fresh_then_clear() { + let thread = "budget-corr-test-lifecycle"; + clear_budget_signal(thread).await; // isolate from other tests + assert!(!has_fresh_budget_signal(thread, BINDING).await); + + record_budget_signal(thread, BINDING).await; + assert!(has_fresh_budget_signal(thread, BINDING).await); + + clear_budget_signal(thread).await; + assert!(!has_fresh_budget_signal(thread, BINDING).await); + } + + #[tokio::test] + async fn stale_signal_is_not_fresh_and_is_evicted() { + let thread = "budget-corr-test-stale"; + // Seed a signal older than the TTL. + record_budget_signal_aged(thread, BINDING, BUDGET_SIGNAL_TTL + Duration::from_secs(1)) + .await; + // Reads as not-fresh and self-evicts. + assert!(!has_fresh_budget_signal(thread, BINDING).await); + // Confirm eviction: still not fresh, and a later in-window seed works. + assert!(!has_fresh_budget_signal(thread, BINDING).await); + record_budget_signal_aged(thread, BINDING, Duration::from_secs(1)).await; + assert!(has_fresh_budget_signal(thread, BINDING).await); + clear_budget_signal(thread).await; + } + + #[tokio::test] + async fn signal_does_not_cross_provider_bindings() { + let thread = "budget-corr-test-binding"; + clear_budget_signal(thread).await; + // Budget hit on the managed route. + record_budget_signal(thread, "openhuman-managed").await; + // A turn re-routed to a different (BYO/local) provider must NOT inherit + // the managed exhaustion — its empty response is unrelated. + assert!(!has_fresh_budget_signal(thread, "byo-deepseek").await); + // The same managed binding still reads fresh (mismatch read above + // evicted it, so re-record to prove the same-binding path). + record_budget_signal(thread, "openhuman-managed").await; + assert!(has_fresh_budget_signal(thread, "openhuman-managed").await); + clear_budget_signal(thread).await; + } + + #[tokio::test] + async fn record_prunes_other_threads_stale_entries() { + let abandoned = "budget-corr-test-abandoned"; + let active = "budget-corr-test-active"; + // An abandoned thread leaves a stale entry behind... + record_budget_signal_aged( + abandoned, + BINDING, + BUDGET_SIGNAL_TTL + Duration::from_secs(1), + ) + .await; + // ...which a later budget event on a DIFFERENT thread sweeps away. + record_budget_signal(active, BINDING).await; + { + let signals = THREAD_BUDGET_SIGNALS.lock().await; + assert!( + !signals.contains_key(abandoned), + "stale entry should be pruned" + ); + assert!(signals.contains_key(active)); + } + clear_budget_signal(active).await; + } +} diff --git a/src/openhuman/channels/providers/web/run_task.rs b/src/openhuman/channels/providers/web/run_task.rs index 47deee5b0..6b5b23fed 100644 --- a/src/openhuman/channels/providers/web/run_task.rs +++ b/src/openhuman/channels/providers/web/run_task.rs @@ -4,7 +4,7 @@ use crate::openhuman::config::rpc as config_rpc; use crate::openhuman::profiles::AgentProfileStore; use crate::openhuman::threads::turn_state::TurnStateStore; -use super::ops::{key_for, THREAD_SESSIONS}; +use super::ops::{key_for, BudgetCorrelation, THREAD_SESSIONS}; use super::progress_bridge::spawn_progress_bridge; use super::session::{ build_session_agent, build_session_fingerprint, normalize_model_override, pick_target_agent_id, @@ -14,7 +14,7 @@ use super::types::SessionEntry; use super::types::{ChatRequestMetadata, WebChatTaskResult}; use super::web_errors::{ classify_inference_error, inference_budget_exceeded_user_message, - is_inference_budget_exceeded_error, + is_empty_provider_response_text, is_inference_budget_exceeded_error, }; #[cfg(any(test, debug_assertions))] @@ -232,6 +232,10 @@ pub(crate) async fn run_chat_task( .await { Ok(response) => { + // A successful turn proves the thread's balance is usable, so drop + // any stale budget-exhausted signal before it could mislabel a + // later genuine empty response. See #3386. + super::ops::clear_budget_signal(thread_id).await; let citations = agent.take_last_turn_citations(); Ok(WebChatTaskResult { full_response: response, @@ -240,19 +244,51 @@ pub(crate) async fn run_chat_task( } Err(err) => { let err_message = err.to_string(); - if is_inference_budget_exceeded_error(&err_message) { - log::warn!( - "[web-channel] inference budget exhausted for client={} thread={} request_id={} error_category=budget_exhausted", - client_id, - thread_id, - request_id - ); - Ok(WebChatTaskResult { - full_response: inference_budget_exceeded_user_message().to_string(), - citations: Vec::new(), - }) + let is_budget = is_inference_budget_exceeded_error(&err_message); + let is_empty = is_empty_provider_response_text(&err_message.to_lowercase()); + // Only consult the cross-turn signal for an empty response that is + // not already a budget error — avoids taking the lock on every turn. + let has_fresh_signal = if !is_budget && is_empty { + super::ops::has_fresh_budget_signal(thread_id, ¤t_fp.provider_binding).await } else { - Err(err_message) + false + }; + match super::ops::classify_budget_correlation(is_budget, is_empty, has_fresh_signal) { + BudgetCorrelation::BudgetExhausted => { + // Remember the exhaustion (scoped to this provider binding) + // so a follow-up empty 200 on this thread+provider (managed + // route closes the SSE clean under credit exhaustion, no + // inline marker) reclassifies as budget. #3386. + super::ops::record_budget_signal(thread_id, ¤t_fp.provider_binding).await; + log::warn!( + "[web-channel] inference budget exhausted for client={} thread={} request_id={} error_category=budget_exhausted", + client_id, + thread_id, + request_id + ); + Ok(WebChatTaskResult { + full_response: inference_budget_exceeded_user_message().to_string(), + citations: Vec::new(), + }) + } + BudgetCorrelation::UpgradeEmptyToBudget => { + // Empty provider response within the budget-signal window: + // the turn most likely failed for the same out-of-credits + // reason but arrived as a clean empty 200 with no budget + // marker. Surface the actionable budget copy. #3386. + log::warn!( + "[web-channel] reclassifying empty provider response as budget-exhausted \ + from recent same-thread signal client={} thread={} request_id={} error_category=budget_exhausted_correlated", + client_id, + thread_id, + request_id + ); + Ok(WebChatTaskResult { + full_response: inference_budget_exceeded_user_message().to_string(), + citations: Vec::new(), + }) + } + BudgetCorrelation::PassThrough => Err(err_message), } } };