From be386040c40dd6dcd246a212bc03b6af87b703dd Mon Sep 17 00:00:00 2001 From: oxoxDev <164490987+oxoxDev@users.noreply.github.com> Date: Fri, 3 Jul 2026 23:51:08 +0530 Subject: [PATCH] fix(api): stop /teams/me/usage flooding from BYO-provider control-plane misroute (#4153) (#4298) Co-authored-by: M3gA-Mind --- src/api/config.rs | 207 +++++++++++++++++++++++++- src/core/jsonrpc.rs | 25 +++- src/core/observability.rs | 35 +++++ src/openhuman/team/ops.rs | 305 +++++++++++++++++++++++++++++++++++++- 4 files changed, 561 insertions(+), 11 deletions(-) diff --git a/src/api/config.rs b/src/api/config.rs index 94f1b9876..a88feffb5 100644 --- a/src/api/config.rs +++ b/src/api/config.rs @@ -154,6 +154,7 @@ pub fn effective_api_url(api_url: &Option) -> String { pub fn effective_backend_api_url(api_url: &Option) -> String { if let Some(u) = non_empty_str(api_url) { let is_local_ai = looks_like_local_ai_endpoint(u); + let is_inference_provider = looks_like_inference_provider_endpoint(u); let is_openhuman = looks_like_openhuman_backend_endpoint(u); // A public third-party inference host (openrouter.ai, api.openai.com, …) // set to its canonical base (`https://openrouter.ai/api/v1`) is neither @@ -172,16 +173,38 @@ pub fn effective_backend_api_url(api_url: &Option) -> String { tracing::debug!( api_url = %redact_url_for_log(u), is_local_ai, + is_inference_provider, is_cloud_inference, is_openhuman, "[api/config] evaluating backend api_url override" ); - // Let the override through only when it is neither a local-AI endpoint - // nor a known cloud inference host, OR when it is one of our own hosted - // backends (user deliberately set `api_url` to + // Let the override through only when it is NOT an inference endpoint + // (local model runner OR remote managed provider), OR when it is one of + // our own hosted backends (user deliberately set `api_url` to // `https://api.tinyhumans.ai/openai/v1/chat/completions`). - if (!is_local_ai && !is_cloud_inference) || is_openhuman { + // + // `config.api_url` doubles as the BYO inference base (see + // `effective_inference_url`), so a user who points it at a managed + // provider (`openrouter.ai`, `api.openmodel.ai`, …) was silently + // sending every CONTROL-PLANE call there too — `/teams/me/usage`, + // `/teams/*`, billing, referral — which the provider answers with a + // 400/404/500. That misroute is the bulk of the `/teams/me/usage` + // non-2xx flood (TAURI-RUST-BSF / -8C / -HDS / -HW1 / -JJ5, GH #4153): + // the `host` tag added in #4058 pinned the destinations to + // `openrouter.ai` / `api.openmodel.ai`, not our backend. The local-AI + // arm already covered Ollama (`OPENHUMAN-TAURI-51 / -80 / -7Z`); this + // widens the same fallback to remote providers. Inference routing is + // unaffected — it resolves through `effective_api_url`, which never + // consults this guard. + // + // `is_cloud_inference` (builtin cloud-provider host check, #4286) is + // kept as an additional signal: it and `is_inference_provider` use + // different detectors (builtin-cloud list vs curated inference domains + + // OpenAI `/v1` path), so either flagging the override as an inference + // base is enough to skip it — strictly safer against control-plane + // misroute. + if (!is_local_ai && !is_inference_provider && !is_cloud_inference) || is_openhuman { let normalized = normalize_backend_api_base_url(u); tracing::trace!( api_url = %redact_url_for_log(u), @@ -194,8 +217,9 @@ pub fn effective_backend_api_url(api_url: &Option) -> String { tracing::debug!( api_url = %redact_url_for_log(u), is_local_ai, + is_inference_provider, is_cloud_inference, - "[api/config] override classified as local AI or cloud inference host — falling back to backend default chain" + "[api/config] override classified as inference endpoint (managed provider or builtin cloud host) — falling back to backend default chain" ); warn_backend_url_fallback_once(u); } @@ -275,6 +299,95 @@ pub fn looks_like_local_ai_endpoint(url: &str) -> bool { port_is_llm || path_is_llm } +/// Well-known managed inference-provider registrable domains. A `config.api_url` +/// pointed at one of these (or a subdomain) is a BYO chat/inference base — never +/// an OpenHuman control-plane backend — so backend calls must NOT route there. +/// +/// Suffix-matched so `api.` / `.` also classify. +/// Kept tight to genuinely managed inference hosts; an unknown custom backend +/// is still honored unless it carries the OpenAI-compatible `/v1` base shape +/// below. `tinyhumans.ai` is deliberately ABSENT — our own hosted backend is +/// recognised by [`looks_like_openhuman_backend_endpoint`] and must route. +const INFERENCE_PROVIDER_DOMAINS: &[&str] = &[ + "openrouter.ai", + "openmodel.ai", + "openai.com", + "anthropic.com", + "groq.com", + "mistral.ai", + "deepseek.com", + "together.ai", + "together.xyz", + "perplexity.ai", + "fireworks.ai", + "deepinfra.com", + "anyscale.com", + "novita.ai", + "hyperbolic.xyz", + "x.ai", + "googleapis.com", + "cohere.ai", + "cohere.com", +]; + +/// Returns `true` when the URL looks like a **remote managed inference +/// provider** base rather than the hosted OpenHuman backend. +/// +/// Complements [`looks_like_local_ai_endpoint`] (which only catches *local* +/// model runners): together they let [`effective_backend_api_url`] fall back to +/// the canonical backend whenever `config.api_url` has been set to an inference +/// base instead of a control-plane base. See the misroute note in +/// `effective_backend_api_url` (GH #4153 / TAURI-RUST-BSF·8C·HDS·HW1·JJ5). +/// +/// Two signals (either is sufficient): +/// 1. **Known provider host** — host equals or is a subdomain of a domain in +/// [`INFERENCE_PROVIDER_DOMAINS`]. +/// 2. **OpenAI-compatible base path** — the path is exactly `/v1` or `/api/v1` +/// (trailing slash ignored). This is the canonical OpenAI-style base and is +/// never an OpenHuman control-plane base. A bare `/v1/chat/completions` is +/// already covered by [`looks_like_local_ai_endpoint`]'s path signal. +/// +/// Our own hosted backend short-circuits to `false` so a user who set +/// `api_url` to `https://api.tinyhumans.ai/...` still reaches the backend. +pub fn looks_like_inference_provider_endpoint(url: &str) -> bool { + let trimmed = url.trim(); + if trimmed.is_empty() { + return false; + } + + // Our own hosted backend is a backend, never a "foreign" inference base. + if looks_like_openhuman_backend_endpoint(trimmed) { + return false; + } + + let parsed = match url::Url::parse(trimmed) { + Ok(u) => u, + Err(_) => return false, + }; + + // ── Signal 1: known managed provider host (apex or subdomain) ────────── + if let Some(host) = parsed.host_str() { + let host = host.to_ascii_lowercase(); + if INFERENCE_PROVIDER_DOMAINS + .iter() + .any(|d| host == *d || host.ends_with(&format!(".{d}"))) + { + return true; + } + } + + // ── Signal 2: OpenAI-compatible bare `/v1` or `/api/v1` base path ────── + // Exact-match both arms (per the doc contract): a longer self-hosted path + // that merely *ends* in `/api/v1` (e.g. `https://backend.internal/svc/api/v1`) + // is a custom backend, not an inference base, and must keep routing. + let path = parsed.path().trim_end_matches('/'); + if path == "/api/v1" || path == "/v1" { + return true; + } + + false +} + /// Returns `true` when the URL's host is one of the known OpenHuman backends. /// /// Used in [`effective_backend_api_url`] to short-circuit the local-AI check: @@ -1133,6 +1246,90 @@ mod tests { assert_eq!(effective_backend_api_url(&None), effective_api_url(&None)); } + // ── GH #4153: remote managed inference providers parked in `api_url` ────── + + #[test] + fn inference_provider_matches_known_remote_hosts() { + // The hosts the #4058 `host` tag pinned the `/teams/me/usage` flood to. + assert!(looks_like_inference_provider_endpoint( + "https://openrouter.ai/api/v1" + )); + assert!(looks_like_inference_provider_endpoint( + "https://api.openmodel.ai/v1" + )); + // Other managed providers, apex and subdomain. + assert!(looks_like_inference_provider_endpoint( + "https://api.openai.com/v1" + )); + assert!(looks_like_inference_provider_endpoint( + "https://api.groq.com/openai/v1" + )); + assert!(looks_like_inference_provider_endpoint( + "https://api.mistral.ai" + )); + } + + #[test] + fn inference_provider_matches_bare_v1_base_on_unknown_host() { + // An unknown OpenAI-compatible provider, recognised by its `/v1` base. + assert!(looks_like_inference_provider_endpoint( + "https://llm.unknown-provider.example/v1" + )); + assert!(looks_like_inference_provider_endpoint( + "https://gw.example.test/api/v1/" + )); + } + + #[test] + fn inference_provider_excludes_openhuman_backend_and_plain_hosts() { + // Our own hosted backend is a backend, even though it serves inference. + assert!(!looks_like_inference_provider_endpoint( + "https://api.tinyhumans.ai/openai/v1/chat/completions" + )); + assert!(!looks_like_inference_provider_endpoint( + "https://staging-api.tinyhumans.ai/" + )); + // A custom self-hosted OpenHuman backend (no provider host, no `/v1` + // base) must keep routing control-plane calls to itself. + assert!(!looks_like_inference_provider_endpoint( + "https://my-openhuman.example.com/" + )); + // Garbage / relative input never panics or matches. + assert!(!looks_like_inference_provider_endpoint("")); + assert!(!looks_like_inference_provider_endpoint("not a url")); + } + + #[test] + fn backend_url_falls_back_for_remote_inference_provider_override() { + // The core of #4153: `config.api_url` set to a managed inference + // provider must NOT be used as the control-plane base; backend calls + // fall back to the canonical default chain. + let _guard = env_lock(); + let _env = EnvSnapshot::clear_backend_env(); + let expected = fallback_backend_base_for_current_build(); + + assert_eq!( + effective_backend_api_url(&Some("https://openrouter.ai/api/v1".to_string())), + expected + ); + assert_eq!( + effective_backend_api_url(&Some("https://api.openmodel.ai/v1".to_string())), + expected + ); + } + + #[test] + fn backend_url_falls_back_to_env_for_remote_inference_provider() { + let _guard = env_lock(); + let _env = EnvSnapshot::clear_backend_env(); + std::env::set_var("BACKEND_URL", "https://staging-api.tinyhumans.ai/"); + + assert_eq!( + effective_backend_api_url(&Some("https://openrouter.ai/api/v1".to_string())), + "https://staging-api.tinyhumans.ai" + ); + } + #[test] fn backend_url_strips_inference_path_from_env() { // Regression: OPENHUMAN-TAURI-H6 / -HN, issue #2075. diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index c864086c2..56bfddae0 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -62,7 +62,7 @@ pub async fn rpc_handler(State(state): State, Json(req): Json ( envelope.message, envelope.data, @@ -121,6 +121,29 @@ pub async fn rpc_handler(State(state): State, Json(req): Json err ({}ms): {}", method, ms, display_message); + } else if crate::core::observability::is_suppressed_usage_probe_backoff( + &display_message, + ) { + // A `/teams/me/usage` probe that the failure-backoff in + // `team::ops` short-circuited within its window — i.e. an + // already-reported repeat. The FIRST failure of the streak + // already hit the backend and reported here normally; demoting + // the repeats is exactly the flood control GH #4153 asks for + // (backpressure, not silent drop). Debug-only, never Sentry. + tracing::debug!( + method = %method, + elapsed_ms = ms as u64, + "[rpc] usage-probe failure-backoff repeat — not reporting to Sentry" + ); + // Keep the internal demotion marker strictly out-of-band: it + // exists only to drive the Sentry-demotion decision above and + // must never reach the RPC client as the error message. Replace + // it with a clean, user-presentable string before the response + // is built below (CodeRabbit on #4153 — don't leak the sentinel). + display_message = + "Usage temporarily unavailable — the last fetch failed and is backing off; \ + it will refresh shortly." + .to_string(); } else if crate::core::observability::is_transient_message_failure(&display_message) { // Downstream call (backend_api / integrations / provider) already // demoted the underlying transient failure to a warn. The error diff --git a/src/core/observability.rs b/src/core/observability.rs index 816d37e30..dfd96a78b 100644 --- a/src/core/observability.rs +++ b/src/core/observability.rs @@ -2848,6 +2848,25 @@ pub fn is_transient_message_failure(msg: &str) -> bool { || contains_transient_transport_phrase(&lower) } +/// Sentinel prefix stamped on a `/teams/me/usage` probe error that the +/// failure-backoff in `crate::openhuman::team::ops` short-circuited — i.e. an +/// already-reported repeat within the backoff window. The FIRST failure of a +/// streak propagates its real error string and reports normally; only the +/// suppressed repeats carry this prefix so the JSON-RPC boundary can demote +/// them (no re-report) instead of re-flooding Sentry. See GH #4153. +/// +/// Single source of truth: the producer (`team::ops::get_usage_with_cache`) +/// builds its sentinel from this constant, and [`is_suppressed_usage_probe_backoff`] +/// matches it — coupled by a unit test so the two cannot drift. +pub const USAGE_PROBE_BACKOFF_PREFIX: &str = "USAGE_PROBE_BACKOFF:"; + +/// Returns true when a message is the usage-probe failure-backoff sentinel +/// (see [`USAGE_PROBE_BACKOFF_PREFIX`]). Anchored on the exact prefix so a real +/// backend error string can never match. +pub fn is_suppressed_usage_probe_backoff(msg: &str) -> bool { + msg.starts_with(USAGE_PROBE_BACKOFF_PREFIX) +} + /// Returns true when a Sentry event is a budget-exhausted 400 that should be /// dropped from `before_send`. /// @@ -3090,6 +3109,22 @@ fn event_contains_budget_exhausted_message(event: &sentry::protocol::Event<'_>) mod tests { use super::*; + #[test] + fn usage_probe_backoff_sentinel_classifies_only_its_own_prefix() { + // The producer in `team::ops` builds its sentinel from this prefix. + let sentinel = format!("{USAGE_PROBE_BACKOFF_PREFIX} recent /teams/me/usage failure"); + assert!(is_suppressed_usage_probe_backoff(&sentinel)); + // Real backend error strings must NEVER match — they keep reporting. + assert!(!is_suppressed_usage_probe_backoff( + "GET /teams/me/usage failed (500 Internal Server Error): " + )); + assert!(!is_suppressed_usage_probe_backoff( + "GET /teams/me/usage failed (404 Not Found); response_body_len=91" + )); + assert!(!is_suppressed_usage_probe_backoff("SESSION_EXPIRED: ...")); + assert!(!is_suppressed_usage_probe_backoff("")); + } + /// Helper must accept `&anyhow::Error`, `&dyn std::error::Error`, and /// plain `&str` — the three shapes that show up at error sites today. #[test] diff --git a/src/openhuman/team/ops.rs b/src/openhuman/team/ops.rs index 76678c3a8..9a7d73444 100644 --- a/src/openhuman/team/ops.rs +++ b/src/openhuman/team/ops.rs @@ -80,12 +80,133 @@ async fn get_authed_value( .map_err(crate::api::flatten_authed_error) } +/// How long a *failed* usage fetch is short-circuited before the backend is +/// probed again. `get_usage` is hammered from two surfaces — the frontend usage +/// poll (the `team_get_usage` RPC) and the pre-call budget gate +/// (`managed_tool_budget_exhausted`) — so a *persistent* non-2xx (a misrouted +/// `BACKEND_URL`, a backend outage) otherwise re-fires on every poll and every +/// managed tool call, re-reporting to Sentry each time. That is the +/// `/teams/me/usage` flood in GH #4153 (TAURI-RUST-BSF/-8C/-HDS/-HW1/-JJ5). +/// +/// This window collapses a persistent fault to ~one backend probe (and so +/// ~one Sentry event) per minute per process while staying responsive: the +/// FIRST failure of a streak still hits the backend and still reports (real +/// signal preserved — backpressure, not silent drop), and any success clears +/// the window immediately. This is defense-in-depth flood control; the actual +/// misroute fix lives in `effective_backend_api_url` (see GH #4153). +const USAGE_FAILURE_BACKOFF: Duration = Duration::from_secs(60); + +/// Process-global anchor of the most recent *reportable* `get_usage` failure. +/// Session-expiry (401) failures are intentionally NOT recorded here — they are +/// handled by their own RPC arm and must keep driving auth recovery. +struct UsageFailureCache { + /// Backend identity + `Instant` of the last reported failure for that + /// backend, if a streak is active. Keying on the backend URL means a + /// changed `config.api_url` (e.g. after the user fixes a misrouted + /// `BACKEND_URL`, or auth/session context that re-points the backend) no + /// longer matches the stored key, so the next probe hits the new backend + /// immediately instead of inheriting the old backend's backoff (GH #4153). + inner: RwLock>, +} + +impl UsageFailureCache { + const fn new() -> Self { + Self { + inner: RwLock::new(None), + } + } + + /// True when a failure for `key` was recorded within `ttl` of `now`. A + /// failure anchored under a different backend key never counts as fresh. + fn is_fresh(&self, key: &str, now: Instant, ttl: Duration) -> bool { + let guard = self.inner.read().unwrap_or_else(|e| e.into_inner()); + guard + .as_ref() + .is_some_and(|(k, at)| k == key && now.duration_since(*at) < ttl) + } + + /// Anchor a fresh failure for `key` (start / keep a streak). + fn record(&self, key: &str, now: Instant) { + let mut guard = self.inner.write().unwrap_or_else(|e| e.into_inner()); + *guard = Some((key.to_string(), now)); + } + + /// Clear the streak — the endpoint recovered. + fn clear(&self) { + let mut guard = self.inner.write().unwrap_or_else(|e| e.into_inner()); + *guard = None; + } +} + +static USAGE_FAILURE_CACHE: UsageFailureCache = UsageFailureCache::new(); + pub async fn get_usage(config: &Config) -> Result, String> { - let data = get_authed_value(config, Method::GET, "/teams/me/usage", None).await?; - Ok(RpcOutcome::single_log( - data, - "team usage fetched from backend", - )) + // Key the failure backoff by the effective backend URL so a failure on one + // backend never suppresses probes after the backend is re-pointed (#4153). + let backend_key = effective_backend_api_url(&config.api_url); + get_usage_with_cache( + &USAGE_FAILURE_CACHE, + &backend_key, + USAGE_FAILURE_BACKOFF, + Instant::now(), + || async { get_authed_value(config, Method::GET, "/teams/me/usage", None).await }, + ) + .await +} + +/// Cache-aware usage fetch. Mirrors the backoff/backpressure shape of +/// [`budget_exhausted_with_cache`] but for *failures*: +/// +/// - Within `ttl` of a recorded failure → short-circuit WITHOUT calling +/// `fetch` (network backpressure) and return the +/// [`crate::core::observability::USAGE_PROBE_BACKOFF_PREFIX`] sentinel, which +/// the JSON-RPC boundary demotes (no re-report). +/// - Otherwise call `fetch`: `Ok` clears the streak; a non-session-expiry `Err` +/// anchors a fresh streak and propagates verbatim (first-of-streak reports); +/// a session-expiry `Err` propagates verbatim WITHOUT anchoring (its own RPC +/// arm handles it / drives auth recovery). +async fn get_usage_with_cache( + cache: &UsageFailureCache, + key: &str, + ttl: Duration, + now: Instant, + fetch: F, +) -> Result, String> +where + F: FnOnce() -> Fut, + Fut: std::future::Future>, +{ + if cache.is_fresh(key, now, ttl) { + tracing::debug!( + "[team] usage probe in failure-backoff window — skipping backend call (suppressing repeat report)" + ); + return Err(format!( + "{} recent /teams/me/usage failure suppressed (backoff)", + crate::core::observability::USAGE_PROBE_BACKOFF_PREFIX + )); + } + + match fetch().await { + Ok(data) => { + cache.clear(); + Ok(RpcOutcome::single_log( + data, + "team usage fetched from backend", + )) + } + Err(err) => { + if crate::core::observability::is_session_expired_message(&err) { + // Session lapse — handled by the session-expired RPC arm and + // drives local session cleanup. Never enter the failure window. + tracing::debug!( + "[team] usage probe failed with session-expiry — not anchoring backoff" + ); + } else { + cache.record(key, now); + } + Err(err) + } + } } fn usage_number(data: &Value, key: &str) -> f64 { @@ -401,6 +522,180 @@ mod tests { assert_eq!(cache.get(Instant::now(), Duration::from_secs(30)), None); } + // ── GH #4153: `/teams/me/usage` failure backoff ────────────────────────── + + use crate::core::observability::is_suppressed_usage_probe_backoff; + + const TTL: Duration = Duration::from_secs(60); + const FAIL_KEY: &str = "backend-under-test"; + + #[test] + fn usage_failure_cache_freshness_window() { + let cache = UsageFailureCache::new(); + let base = Instant::now(); + assert!( + !cache.is_fresh(FAIL_KEY, base, TTL), + "empty cache is never fresh" + ); + cache.record(FAIL_KEY, base); + assert!(cache.is_fresh(FAIL_KEY, base + Duration::from_secs(5), TTL)); + assert!(!cache.is_fresh(FAIL_KEY, base + Duration::from_secs(61), TTL)); + cache.clear(); + assert!( + !cache.is_fresh(FAIL_KEY, base, TTL), + "cleared cache is never fresh" + ); + } + + // GH #4153: a failure anchored under one backend key must NOT suppress a + // probe for a different backend (e.g. after the user fixes BACKEND_URL or + // the session re-points the backend) — otherwise the new route never gets + // tested for up to the backoff window. + #[test] + fn failure_backoff_is_keyed_per_backend() { + let cache = UsageFailureCache::new(); + let base = Instant::now(); + cache.record("https://old.example/api/v1", base); + assert!(cache.is_fresh("https://old.example/api/v1", base, TTL)); + assert!( + !cache.is_fresh("https://new.example/api/v1", base, TTL), + "a different backend key must not inherit the old backend's backoff" + ); + } + + // T1 — first failure of a streak hits the backend, reports verbatim, anchors. + #[tokio::test] + async fn first_failure_reports_and_anchors() { + let cache = UsageFailureCache::new(); + let base = Instant::now(); + let calls = AtomicUsize::new(0); + let err = get_usage_with_cache(&cache, FAIL_KEY, TTL, base, || { + calls.fetch_add(1, Ordering::SeqCst); + async { Err("GET /teams/me/usage failed (500 Internal Server Error): ".to_string()) } + }) + .await + .unwrap_err(); + assert_eq!(calls.load(Ordering::SeqCst), 1, "backend probed once"); + assert!( + !is_suppressed_usage_probe_backoff(&err), + "first failure must NOT be the backoff sentinel (it reports): {err}" + ); + assert!(cache.is_fresh(FAIL_KEY, base, TTL), "streak anchored"); + } + + // T2 — a repeat inside the window short-circuits WITHOUT touching the + // backend and returns the demote sentinel. + #[tokio::test] + async fn repeat_within_window_suppressed_and_skips_backend() { + let cache = UsageFailureCache::new(); + let base = Instant::now(); + cache.record(FAIL_KEY, base); + let calls = AtomicUsize::new(0); + let err = + get_usage_with_cache(&cache, FAIL_KEY, TTL, base + Duration::from_secs(5), || { + calls.fetch_add(1, Ordering::SeqCst); + async { panic!("fetch must not run inside the backoff window") } + }) + .await + .unwrap_err(); + assert_eq!( + calls.load(Ordering::SeqCst), + 0, + "backend NOT probed (backpressure)" + ); + assert!( + is_suppressed_usage_probe_backoff(&err), + "repeat must carry the backoff sentinel: {err}" + ); + } + + // T3 — once the window expires the backend is probed again (≤1 report/min). + #[tokio::test] + async fn window_expiry_reprobes_and_reports() { + let cache = UsageFailureCache::new(); + let base = Instant::now(); + cache.record(FAIL_KEY, base); + let later = base + Duration::from_secs(61); + let calls = AtomicUsize::new(0); + let err = get_usage_with_cache(&cache, FAIL_KEY, TTL, later, || { + calls.fetch_add(1, Ordering::SeqCst); + async { Err("GET /teams/me/usage failed (500 Internal Server Error): ".to_string()) } + }) + .await + .unwrap_err(); + assert_eq!(calls.load(Ordering::SeqCst), 1, "stale window re-probes"); + assert!( + !is_suppressed_usage_probe_backoff(&err), + "re-probe failure reports" + ); + assert!( + cache.is_fresh(FAIL_KEY, later, TTL), + "streak re-anchored at the new probe" + ); + } + + // T4 — a success clears the streak so the next failure reports immediately. + #[tokio::test] + async fn success_clears_streak() { + let cache = UsageFailureCache::new(); + let base = Instant::now(); + cache.record(FAIL_KEY, base); + let later = base + Duration::from_secs(61); // stale → fetch runs + let outcome = get_usage_with_cache(&cache, FAIL_KEY, TTL, later, || async { + Ok(serde_json::json!({"remainingUsd": 5.0})) + }) + .await + .expect("success"); + assert_eq!(outcome.value["remainingUsd"], 5.0); + assert!( + !cache.is_fresh(FAIL_KEY, later, TTL), + "success cleared the streak" + ); + } + + // T5 — session-expiry must flow verbatim and must NOT anchor the window + // (its own RPC arm drives auth recovery). + #[tokio::test] + async fn session_expiry_bypasses_backoff() { + let cache = UsageFailureCache::new(); + let base = Instant::now(); + let err = get_usage_with_cache(&cache, FAIL_KEY, TTL, base, || async { + Err( + "SESSION_EXPIRED: backend rejected session token on GET /teams/me/usage" + .to_string(), + ) + }) + .await + .unwrap_err(); + assert!( + err.contains("SESSION_EXPIRED"), + "propagated verbatim: {err}" + ); + assert!(!is_suppressed_usage_probe_backoff(&err)); + assert!( + !cache.is_fresh(FAIL_KEY, base, TTL), + "session-expiry must not start a backoff streak" + ); + } + + // T6 — the producer's sentinel and the classifier are coupled (no drift). + #[tokio::test] + async fn produced_sentinel_is_classified() { + let cache = UsageFailureCache::new(); + let base = Instant::now(); + cache.record(FAIL_KEY, base); + let err = get_usage_with_cache(&cache, FAIL_KEY, TTL, base, || async { + unreachable!("fresh window short-circuits") + }) + .await + .unwrap_err(); + assert!( + err.starts_with(crate::core::observability::USAGE_PROBE_BACKOFF_PREFIX), + "sentinel built from the shared prefix constant: {err}" + ); + assert!(is_suppressed_usage_probe_backoff(&err)); + } + #[test] fn budget_cache_returns_value_within_ttl_and_expires_after() { let cache = BudgetProbeCache::new();