Hold a 402, report a discarded api_url, and sweep gateway drift

CHANGELOG.md

@@ -15,6 +15,11 @@ All notable changes to this project will be documented here. Format follows [Kee
 
 ### Fixed
 
+- **A `402` from anything in front of Lago silently discarded every billable event for the whole outage.** `402` was classified permanent, which routes a batch to `_send_individually`; every isolated send then `402`s too and is logged and dropped for good. Measured against a server returning 402: **5 events in, 6 HTTP calls out, 0 recoverable**, with a single `on_error` for the lot. The classification test is "is what makes this fail a property of the BATCH?" — true for 413 (size) and 415 (media type), which are constant across retries, and false for 402: *payment required* is a property of the **account** and stops being true the moment someone pays, exactly like the 429 that was deliberately carved out. `402` is now transient, so a lapsed account holds and retries instead — bounded, oldest-first, reported, and fully recoverable inside the buffer window. Also verified live against a real Lago that **none of 402/413/415 is reachable from Lago itself** (its surface is 400/401/403/404/405/422/429): an oversized batch answers **422 `too_many_events`**, which already routes to the split path, and a duplicate `transaction_id` answers **422 `value_already_exist`**, not the 409 the comment claimed. 413/415 are kept for an intermediary (nginx's `client_max_body_size` really does answer 413 before the request reaches Rails) and the comments now say so.
+- **An explicitly-passed falsy `api_url` silently resolved to PRODUCTION Lago.** Preferring the config value over `""` is right — `requests` raises `MissingSchema`, which is not a `LagoApiError`, so the queue retried forever and all billing stopped — but `LagoConfig`'s default is the production URL, so `api_url=os.environ.get("LAGO_API_URL", "")` with the var unset resolved to production with **no `on_error` and no log**. Verified live: it resolves to production silently. For a CI job or a developer holding a real production key that writes live billing data, and ingested events cannot be un-ingested. The fallback is unchanged (so the original clobber bug stays fixed) but is now reported under `config.api_url` through the same log-plus-callback floor as every other drop path.
+- **`usage_metadata` from the Cloudflare gateway got no drift sweep, and had already lost two counters.** `extras` was a fixed three-key dict, so any counter the adapter does not map vanished with no error and no `on_error` — the one place violating the drift contract `adapters/openai_native.py` enforces. Not hypothetical: a live Logs API pull found **`neurons`** (Cloudflare's Workers AI billing unit) and **`units`** (a cost quantity) dropped on every row, and **`units` appears in none of the 14 captured fixtures** — the hand-maintained key enumeration in the module docstring had already drifted past reality, which is exactly the failure mode a snapshot invites. Unmapped keys are now swept into `extras["usage_metadata"]` against an explicit `_MAPPED_USAGE_KEYS` set, so a ninth spelling surfaces on its own instead of needing another 14-fixture audit. Deliberately **nested** rather than merged flat into `extras`: the poller reads `extras["cached"]` to decide whether to skip billing a request Cloudflare served for free, so a future `usage_metadata` key called `cached` or `step` must not be able to shadow it.
+- **`prime()` re-downloaded the ~400-model OpenRouter catalogue regardless of the TTL.** It set `_openrouter_stale` unconditionally, and it is called from `_auto_prime_pricing_for` on a matching `wrap()` and from `warm_pricing()` — both of which a server can run per request — so `pricing_ttl_seconds` never applied on that path at all. Measured against the live catalogue with the shipped 1-hour TTL: **4 `prime()`+`maybe_refresh()` cycles produced 4 full downloads where 1 was correct; now 1.** Gated on the same "no table, or past the TTL" test `lookup()` already uses, so priming and looking up cannot disagree about what needs fetching.
+- **A failed pricing fetch retried every tick, forever, with no backoff, ahead of the drain.** Only the success path cleared a source's stale flag, so a bad credential re-attempted on every queue tick, each attempt costing up to the 10s `_get_json` timeout, all of it before the drain. Measured with a bad Cloudflare token: **5 ticks produced 5 real requests and 5 `on_error` reports; now 1**, and it still recovers once the window expires (verified: throttled during the backoff, retried after it). Per-source 1→2→4→…→60s backoff, matching the queue's own send backoff, so one bad credential cannot delay the three healthy tables. The four fetches deliberately stay **sequential** here: this refresh runs on a blocking daemon thread, where introducing a thread pool for four fetches would be a larger change than the problem warrants. JS additionally parallelises them under `Promise.allSettled`, which its event loop makes free — a language-inherent divergence, like `os.register_at_fork` vs `AsyncLocalStorage`.
 - **A recovery path silently reversed FIFO order.** `_send_individually` re-queued each transiently-failing event as it went, and `_replay_failed` PREPENDS — so a 413 batch of `a,b,c,d,e` whose `b,c,d` failed while isolated came back as `d,c,b`. FIFO is the queue's contract: it is what makes the oldest-dropped-first overflow policy and Lago's own event ordering mean anything. Survivors are now collected and re-queued once. Present in **both** repos, not just JS as first reported.
 - **A negative token count was dropped without a word.** `nonzero_numeric` correctly filters it (Lago would otherwise sum a negative billable quantity), but this was the last drop path in the SDK that never reached `on_error` — the same gap already closed for queue overflow and for an unresolvable subscription. It is reachable, not theoretical: `CanonicalUsage` is exported and `emit()` takes one directly, which is the documented way to backfill usage the SDK did not intercept, so a caller computing a delta wrongly really can hand us one. Now reported under `negative_tokens`, before the empty-check, so an event whose only fields were negative still reports instead of returning silently.
 - **One log line per dropped event, not two.** `_report_error` already invokes `on_error` AND logs; an extra `logger.error` beside it emitted the same drop twice at two levels, so a customer grepping logs counted one lost call as two. The JS port logged **nothing** for the same event, so the two repos reported 2 lines vs 0 — `reportError` there now logs as well, since `onError` is opt-in and the log is the floor.

src/lago_agent_sdk/gateway/adapters/cloudflare_gateway.py

@@ -14,12 +14,18 @@
 
 Cloudflare reports its OWN counter vocabulary here, not the provider's. Across all
 14 captured fixtures — Anthropic, Workers AI, Mistral and Gemini, via every ingress
-method — the only keys that ever appear are `input_tokens`, `output_tokens`,
-`total_tokens`, `input_cached_tokens`, `input_cache_creation_tokens`, `neurons`,
-`input_text_tokens` and `reasoningTokens`. Not one provider-native key shows up:
-no Anthropic `cache_read_input_tokens`, no Gemini `thoughtsTokenCount` or
+method — the keys that appear are `input_tokens`, `output_tokens`, `total_tokens`,
+`input_cached_tokens`, `input_cache_creation_tokens`, `neurons`, `input_text_tokens`
+and `reasoningTokens`. Not one provider-native key shows up: no Anthropic
+`cache_read_input_tokens`, no Gemini `thoughtsTokenCount` or
 `cachedContentTokenCount`.
 
+That list is a snapshot and has already been overtaken once: a live Logs API pull
+also returned `units`, which appears in none of the fixtures. Treat the enumeration
+as illustrative, not exhaustive — `_MAPPED_USAGE_KEYS` plus the drift sweep into
+`extras["usage_metadata"]` is what actually keeps an unrecognized counter from being
+lost, and it needs no re-audit to stay correct.
+
 That vocabulary is *mostly* snake_case, with `reasoningTokens` as a camelCase
 outlier — Cloudflare's own inconsistency, not a provider key leaking through
 (Gemini's native spelling for the same quantity is `thoughtsTokenCount`, which
@@ -62,6 +68,47 @@ def _safe_str(v: Any) -> str:
     return v if isinstance(v, str) else ""
 
 
+# Every `usage_metadata` spelling this adapter accounts for: the ones `_first_int`
+# consults below, plus the three that are redundant with the top-level `tokens_in` /
+# `tokens_out` the adapter reads directly. Anything NOT in here is swept into
+# `extras["usage_metadata"]` rather than dropped — see the drift note on `extras`.
+#
+# Keep this in sync with the `_first_int` calls. It is the mechanism that makes the
+# module docstring's key enumeration self-maintaining instead of a hand-audited
+# snapshot: a spelling nobody has seen shows up in `extras` on its own.
+_MAPPED_USAGE_KEYS = frozenset(
+    {
+        # cache_read
+        "input_cached_tokens",
+        "inputCachedTokens",
+        "cachedContentTokenCount",
+        "cache_read_input_tokens",
+        # cache_write
+        "input_cache_creation_tokens",
+        "inputCacheCreationTokens",
+        "cache_creation_input_tokens",
+        # reasoning
+        "reasoningTokens",
+        "reasoning_tokens",
+        "thoughtsTokenCount",
+        # Read from the top level instead, so not drift when they appear here.
+        "input_tokens",
+        "output_tokens",
+        "total_tokens",
+    }
+)
+
+
+def _unmapped_usage(usage_meta: dict[str, Any]) -> dict[str, Any]:
+    """Any `usage_metadata` key this adapter does not account for, wrapped for `extras`.
+
+    Returns an empty dict when everything was recognized, so the key is absent rather
+    than present-and-empty in the overwhelmingly common case.
+    """
+    unmapped = {k: v for k, v in usage_meta.items() if k not in _MAPPED_USAGE_KEYS}
+    return {"usage_metadata": unmapped} if unmapped else {}
+
+
 def _first_int(meta: dict[str, Any], *names: str) -> int:
     """First of `names` present in `meta` with a usable value, as an int.
 
@@ -173,6 +220,23 @@ def extract_cloudflare_log(entry: dict[str, Any]) -> CanonicalUsage:
             "cached": entry.get("cached"),
             "step": entry.get("step"),
             "log_id": entry.get("id"),
+            # Drift sweep — the same contract `adapters/openai_native.py` enforces, and
+            # for the same reason: a counter this adapter does not map must not vanish
+            # without an error or an on_error. `extras` used to be exactly the three
+            # keys above, so `usage_metadata` got no sweep at all, and that was not
+            # hypothetical — a live Logs API pull found `neurons` (Cloudflare's Workers
+            # AI billing unit) and `units` (a cost quantity) being dropped on every row,
+            # and `units` appears in NO captured fixture, so the hand-maintained
+            # enumeration had already drifted past what this file claimed to know. A
+            # money-relevant counter going missing this way surfaces first as a
+            # reconciliation gap, not as a failure.
+            #
+            # NESTED, not merged flat into `extras`: the poller reads `extras["cached"]`
+            # to decide whether to skip billing a request Cloudflare served for free, so
+            # a future `usage_metadata` key called `cached` or `step` must not be able to
+            # shadow it. Omitted entirely when there is no drift, to keep the common case
+            # identical to what callers already see.
+            **_unmapped_usage(usage_meta),
         },
     )
 

src/lago_agent_sdk/pricing.py

@@ -55,6 +55,11 @@
 
 logger = logging.getLogger("lago_agent_sdk.pricing")
 
+# Ceiling for a pricing source's post-failure backoff — the same 60s cap the event
+# queue uses for send retries, so a persistently-broken credential settles into one
+# attempt a minute instead of one per flush tick.
+_MAX_PRICING_BACKOFF_SECONDS = 60.0
+
 OPENROUTER_URL = "https://openrouter.ai/api/v1/models"
 AWS_PRICING_HOST = "https://pricing.us-east-1.amazonaws.com"
 AWS_BEDROCK_REGION_INDEX = f"{AWS_PRICING_HOST}/offers/v1.0/aws/AmazonBedrock/current/region_index.json"
@@ -1005,6 +1010,9 @@ def __init__(
         # ever requiring a separate LagoConfig.mistral_api_key.
         self._mistral_api_key_override: str | None = None
         self._refreshing: set[str] = set()
+        # Per-source post-failure backoff — see `_in_backoff`.
+        self._failure_backoff_until: dict[str, float] = {}
+        self._failure_backoff_seconds: dict[str, float] = {}
 
     def _heal_fork(self) -> None:
         """Self-heal after a fork: a lock copied from the parent may be held by a
@@ -1046,14 +1054,61 @@ def prime(self, providers: Iterable[str] = ()) -> None:
         cold-start cost. Unknown provider names are silently ignored (no
         source is warmed) rather than raising, since this is a hint, not a
         contract."""
+        # Gated on "is this table actually cold?", NOT unconditional. `prime()` is called
+        # from `_auto_prime_pricing_for` on a matching `wrap()` and from `warm_pricing()`,
+        # both of which a server can run per request — and flagging an in-TTL table stale
+        # meant the ~400-model OpenRouter catalogue was re-downloaded on essentially every
+        # flush tick, so `pricing_ttl_seconds` never applied on this path at all. Measured
+        # against the live catalogue with the shipped 1-hour TTL: 4 prime()+maybe_refresh()
+        # cycles produced 4 full downloads where 1 was correct.
+        #
+        # "Cold" is the same test `lookup()` already uses — no table, or past the TTL — so
+        # priming and looking up cannot disagree about what needs fetching.
         with self._lock:
-            self._openrouter_stale = True
+            if self._is_cold(self._openrouter, self._openrouter_fetched):
+                self._openrouter_stale = True
             for p in providers:
                 key = (p or "").lower()
                 if key == "workers-ai":
-                    self._cloudflare_stale = True
+                    if self._is_cold(self._cloudflare_workers_ai, self._cloudflare_fetched):
+                        self._cloudflare_stale = True
                 elif key == "mistral":
-                    self._mistral_stale = True
+                    if self._is_cold(self._mistral_aliases, self._mistral_fetched):
+                        self._mistral_stale = True
+
+    def _is_cold(self, table: Any, fetched_at: float) -> bool:
+        """True when a table needs fetching: absent, or older than the TTL.
+
+        Caller must hold `self._lock`.
+        """
+        return table is None or (time.time() - fetched_at) >= self._ttl
+
+    def _in_backoff(self, source: str) -> bool:
+        """True while `source` is inside its post-failure backoff window.
+
+        A failed fetch used to leave its stale flag set and nothing else, so the next tick
+        retried immediately — every tick, forever, with no delay, each attempt costing up
+        to the 10s `_get_json` timeout, all of it on the queue thread AHEAD of the drain.
+        Measured with a bad Cloudflare token: 4 ticks produced 4 real requests and 4
+        `on_error` reports.
+
+        Same 1→2→4→…→60s shape as the queue's own send backoff, tracked per source so one
+        bad credential cannot delay the three healthy tables.
+        """
+        with self._lock:
+            return time.time() < self._failure_backoff_until.get(source, 0.0)
+
+    def _note_failure(self, source: str) -> None:
+        with self._lock:
+            prev = self._failure_backoff_seconds.get(source, 0.0)
+            nxt = 1.0 if prev == 0.0 else min(prev * 2, _MAX_PRICING_BACKOFF_SECONDS)
+            self._failure_backoff_seconds[source] = nxt
+            self._failure_backoff_until[source] = time.time() + nxt
+
+    def _note_success(self, source: str) -> None:
+        with self._lock:
+            self._failure_backoff_seconds.pop(source, None)
+            self._failure_backoff_until.pop(source, None)
 
     def learn_mistral_api_key(self, api_key: str) -> None:
         """Adopt a Mistral API key discovered from a wrapped client, so
@@ -1129,16 +1184,36 @@ def maybe_refresh(self) -> None:
         ):
             return
         with self._lock:
-            do_openrouter = self._openrouter_stale and "openrouter" not in self._refreshing
+            now = time.time()
+
+            def _ready(source: str) -> bool:
+                # Inlined rather than calling `_in_backoff`, which takes the lock we hold.
+                return now >= self._failure_backoff_until.get(source, 0.0)
+
+            do_openrouter = (
+                self._openrouter_stale and "openrouter" not in self._refreshing and _ready("openrouter")
+            )
             if do_openrouter:
                 self._refreshing.add("openrouter")
-            do_cloudflare = self._cloudflare_stale and "cloudflare_workers_ai" not in self._refreshing
+            do_cloudflare = (
+                self._cloudflare_stale
+                and "cloudflare_workers_ai" not in self._refreshing
+                and _ready("cloudflare_workers_ai")
+            )
             if do_cloudflare:
                 self._refreshing.add("cloudflare_workers_ai")
-            do_mistral = self._mistral_stale and "mistral_aliases" not in self._refreshing
+            do_mistral = (
+                self._mistral_stale
+                and "mistral_aliases" not in self._refreshing
+                and _ready("mistral_aliases")
+            )
             if do_mistral:
                 self._refreshing.add("mistral_aliases")
-            regions = [r for r in self._bedrock_stale if f"bedrock:{r}" not in self._refreshing]
+            regions = [
+                r
+                for r in self._bedrock_stale
+                if f"bedrock:{r}" not in self._refreshing and _ready(f"bedrock:{r}")
+            ]
             for r in regions:
                 self._refreshing.add(f"bedrock:{r}")
 
@@ -1149,7 +1224,9 @@ def maybe_refresh(self) -> None:
                     self._openrouter = table
                     self._openrouter_fetched = time.time()
                     self._openrouter_stale = False
+                self._note_success("openrouter")
             except Exception as exc:  # noqa: BLE001
+                self._note_failure("openrouter")
                 self._report(exc, "pricing.fetch_openrouter")
             finally:
                 with self._lock:
@@ -1162,7 +1239,9 @@ def maybe_refresh(self) -> None:
                     self._cloudflare_workers_ai = table_cf
                     self._cloudflare_fetched = time.time()
                     self._cloudflare_stale = False
+                self._note_success("cloudflare_workers_ai")
             except Exception as exc:  # noqa: BLE001
+                self._note_failure("cloudflare_workers_ai")
                 self._report(exc, "pricing.fetch_cloudflare_workers_ai")
             finally:
                 with self._lock:
@@ -1177,7 +1256,9 @@ def maybe_refresh(self) -> None:
                     self._mistral_aliases = aliases
                     self._mistral_fetched = time.time()
                     self._mistral_stale = False
+                self._note_success("mistral_aliases")
             except Exception as exc:  # noqa: BLE001
+                self._note_failure("mistral_aliases")
                 self._report(exc, "pricing.fetch_mistral_aliases")
             finally:
                 with self._lock:
@@ -1190,7 +1271,9 @@ def maybe_refresh(self) -> None:
                     self._bedrock[r] = table
                     self._bedrock_fetched[r] = time.time()
                     self._bedrock_stale.discard(r)
+                self._note_success(f"bedrock:{r}")
             except Exception as exc:  # noqa: BLE001
+                self._note_failure(f"bedrock:{r}")
                 self._report(exc, "pricing.fetch_bedrock")
             finally:
                 with self._lock: