From fc62f4d2d1f468e8f80633fd49aff3d9d5d6cacc Mon Sep 17 00:00:00 2001 From: Anass Date: Thu, 20 Aug 2026 16:51:11 +0200 Subject: [PATCH] Hold a 402, report a discarded api_url, and sweep gateway drift MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A 402 from anything in front of Lago silently discarded every billable event for the whole outage. It was classified permanent, which routes a batch to _send_individually; every isolated send then 402s 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 test for the permanent set is "is what makes this fail a property of the BATCH?". True for 413 (size) and 415 (media type), which are constant across retries. False for 402: payment required is a property of the ACCOUNT and stops being true the moment someone pays — the same shape as the 429 that was deliberately carved out, recoverable by an out-of-band change rather than by sending something different. 402 is now transient, so a lapsed account holds and retries: bounded, oldest-first, reported, and recoverable inside the buffer window. Probing a real Lago also showed 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 as its exemplar. 413/415 stay for an intermediary, since nginx's client_max_body_size really does answer 413 before the request reaches Rails, and the comments now say which reader they are for. An explicitly-passed falsy api_url silently resolved to PRODUCTION. Preferring the config value over "" is right, because 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. 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; it 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. 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 contract test_drift.py enforces for the native adapters. 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, so the hand-maintained enumeration in the module docstring had already drifted past reality. Unmapped keys now sweep 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. Nested rather than merged flat, because the poller reads extras["cached"] to decide whether to skip billing a request Cloudflare served for free, and 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. 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 tick, each attempt costing up to the 10s _get_json timeout. 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. 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 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 against AsyncLocalStorage. Every fix has a test that fails when the fix is reverted. Green at this commit: ruff, ruff format, mypy strict, 552 unit tests. Each finding was also confirmed against live infrastructure before and after the change — a real local Lago, the real OpenRouter catalogue, the real Cloudflare model and Logs APIs. --- CHANGELOG.md | 5 + .../gateway/adapters/cloudflare_gateway.py | 72 +++++++++- src/lago_agent_sdk/pricing.py | 97 +++++++++++++- src/lago_agent_sdk/queue.py | 52 +++++--- src/lago_agent_sdk/sdk.py | 21 +++ .../adapters/test_cloudflare_gateway.py | 124 ++++++++++++++++++ tests/unit/test_queue.py | 47 ++++++- 7 files changed, 389 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fd388fa..186b3f4 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/lago_agent_sdk/gateway/adapters/cloudflare_gateway.py b/src/lago_agent_sdk/gateway/adapters/cloudflare_gateway.py index 8b4d21f..0a0236a 100644 --- a/src/lago_agent_sdk/gateway/adapters/cloudflare_gateway.py +++ b/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), }, ) diff --git a/src/lago_agent_sdk/pricing.py b/src/lago_agent_sdk/pricing.py index 7475d8f..009b080 100644 --- a/src/lago_agent_sdk/pricing.py +++ b/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: diff --git a/src/lago_agent_sdk/queue.py b/src/lago_agent_sdk/queue.py index 18c59d0..f48cd42 100644 --- a/src/lago_agent_sdk/queue.py +++ b/src/lago_agent_sdk/queue.py @@ -33,22 +33,44 @@ # Statuses where re-sending the SAME batch can never succeed: the request itself # is the problem (malformed body, bad credentials, a transaction_id Lago has -# already accepted). Deliberately an explicit list rather than the 400-499 range, -# because two 4xx statuses mean "try again, later": 429 (rate limited) and 408 -# (request timeout). Treating those as permanent dropped billable events AND fanned -# one throttled batch out into up to `max_batch_size` extra requests aimed at the -# server that had just asked us to slow down. +# already accepted — which Lago reports as 422 `value_already_exist`, verified live, +# NOT as 409; 409 is in the set only as defence against an intermediary that uses it). +# Deliberately an explicit list rather than the 400-499 range, because two 4xx statuses +# mean "try again, later": 429 (rate limited) and 408 (request timeout). Treating those +# as permanent dropped billable events AND fanned one throttled batch out into up to +# `max_batch_size` extra requests aimed at the server that had just asked us to slow +# down. # -# 413/402/415 are in the set for the OPPOSITE reason to 429: re-sending the same batch -# provably cannot succeed (too large, payment required, wrong media type), so treating -# them as transient re-prepended the identical batch at the head of the FIFO and backed -# off to 60s forever, blocking every event behind it until the buffer overflowed. Being -# "permanent" here routes them to `_send_individually`, which SPLITS the batch and -# delivers what is deliverable — so a 413 on a 100-event batch becomes 100 single-event -# sends rather than a stalled queue. That is the behaviour we want, and the batch that -# most needs splitting was the one that never reached it. 405/410 stay transient: they -# usually indicate a misrouted or retired endpoint, which a deploy can fix. -_PERMANENT_STATUSES = frozenset({400, 401, 402, 403, 404, 409, 413, 415, 422}) +# 413/415 are in the set for the OPPOSITE reason to 429: re-sending the same batch +# provably cannot succeed, because what makes it fail is a property OF THE BATCH (its +# size, its media type) and that is constant across retries. Treating them as transient +# re-prepended the identical batch at the head of the FIFO and backed off to 60s +# forever, blocking every event behind it until the buffer overflowed. Being "permanent" +# here routes them to `_send_individually`, which SPLITS the batch and delivers what is +# deliverable — so a 413 on a 100-event batch becomes 100 single-event sends rather than +# a stalled queue. +# +# Neither is reachable from Lago itself: its API surface is 400/401/403/404/405/422/429, +# and an oversized batch comes back 422 `too_many_events` (verified live against a real +# instance) which already routes to the split path. They are kept for an intermediary in +# front of Lago — nginx's `client_max_body_size` genuinely does answer 413 without the +# request ever reaching Rails. +# +# 402 was in this set and is NOT, deliberately. It fails the same test: "payment +# required" is a property of the ACCOUNT, not of the batch, so it stops being true the +# moment someone pays — the same shape as 429, recoverable by an out-of-band change +# rather than by sending something different. Classified permanent it was actively +# destructive: the batch 402s, routes to `_send_individually`, every isolated send 402s +# too, and each one is logged and dropped, so a lapsed account silently discarded every +# billable event for the whole outage with one `on_error` for the lot. Measured against a +# server returning 402: 5 events in, 6 HTTP calls out, 0 recoverable. As transient they +# are held and retried instead — a lapsed account head-of-line-blocks at the 60s ceiling +# until `max_buffer_size` overflows, which is bounded, oldest-first and reported, and +# fully recoverable if the account is fixed inside the buffer window. +# +# 405/410 stay transient: they usually indicate a misrouted or retired endpoint, which a +# deploy can fix. +_PERMANENT_STATUSES = frozenset({400, 401, 403, 404, 409, 413, 415, 422}) def _is_permanent_failure(exc: Exception) -> bool: diff --git a/src/lago_agent_sdk/sdk.py b/src/lago_agent_sdk/sdk.py index c9edfbd..1a0a584 100644 --- a/src/lago_agent_sdk/sdk.py +++ b/src/lago_agent_sdk/sdk.py @@ -70,6 +70,8 @@ def __init__( # classifies it transient, re-prepends the batch and retries at the 60s ceiling # forever. All billing stops, nothing is dropped or escalated, and the only # symptom is a growing buffer. + # + # Falling back is right, but it must not be SILENT — see the report below. if api_url: self.config.api_url = api_url if default_subscription_id is not None: @@ -77,6 +79,25 @@ def __init__( if verify_ssl is not None: self.config.verify_ssl = verify_ssl + # A caller who passed `api_url` explicitly MEANT to point somewhere specific. + # Discarding a falsy one is the safe choice for delivery, but doing it silently + # is the one outcome that must not happen here: `LagoConfig`'s default is + # PRODUCTION, so `api_url=os.environ.get("LAGO_API_URL", "")` with the var unset + # now resolves to production Lago and every event is accepted. For a CI job or a + # developer holding a real production key that writes live billing data, and + # ingested events cannot be un-ingested. `on_error` is opt-in, so this reports + # through the same log-plus-callback floor as every other drop path rather than + # trusting a callback to exist. + if api_url is not None and not api_url: + self._report_error( + ValueError( + f"api_url was explicitly set to an empty value; falling back to " + f"{self.config.api_url}. Set LAGO_API_URL (or pass config.api_url) " + f"if you did not intend to send events there." + ), + "config.api_url", + ) + self._lago_client = LagoClient( api_key=self.config.api_key, api_url=self.config.api_url, diff --git a/tests/unit/gateway/adapters/test_cloudflare_gateway.py b/tests/unit/gateway/adapters/test_cloudflare_gateway.py index 1e6c8cc..bad7436 100644 --- a/tests/unit/gateway/adapters/test_cloudflare_gateway.py +++ b/tests/unit/gateway/adapters/test_cloudflare_gateway.py @@ -462,3 +462,127 @@ def test_provider_native_cache_and_reasoning_spellings_are_accepted() -> None: } ) assert both.cache_read == 11 + + +# ---------------------------------------------------------------------- +# Drift contract for `usage_metadata`. +# +# `extras` used to be a fixed three-key dict, so any counter this adapter did not map +# was silently dropped. That was not hypothetical: a live Logs API pull found `neurons` +# and `units` vanishing on every row, and `units` appears in no captured fixture — the +# hand-maintained enumeration in the module docstring had already drifted past reality. +# Same contract `test_drift.py` pins for the native adapters. +# ---------------------------------------------------------------------- +def test_drift_keeps_an_unmapped_counter_instead_of_dropping_it() -> None: + """Exactly the shape seen live (entry 01M0FEZ2Y7QMQR1HT11GVT2HCE).""" + u = extract_cloudflare_log( + { + "id": "log_1", + "cached": False, + "step": 0, + "tokens_in": 37, + "tokens_out": 2, + "provider": "workers-ai", + "model": "@cf/meta/llama-3.3-70b-instruct-fp8-fast", + "usage_metadata": { + "input_tokens": 37, + "output_tokens": 2, + "total_tokens": 39, + "input_cached_tokens": 0, + "neurons": 1.396314412355423, + "units": 0.00001535945853590965, + }, + } + ) + # Cloudflare's Workers AI billing unit, and a cost quantity — both money-relevant. + assert u.extras["usage_metadata"] == { + "neurons": 1.396314412355423, + "units": 0.00001535945853590965, + } + + +def test_drift_sweeps_a_counter_nobody_has_ever_seen() -> None: + u = extract_cloudflare_log( + { + "tokens_in": 10, + "tokens_out": 1, + "provider": "anthropic", + "usage_metadata": {"input_tokens": 10, "audio_input_tokens": 512}, + } + ) + assert u.extras["usage_metadata"] == {"audio_input_tokens": 512} + + +def test_drift_never_shadows_the_pollers_own_billing_inputs() -> None: + """`extras["cached"]` decides whether to skip billing a request Cloudflare served + for free. A usage_metadata key of the same name must not be able to overwrite it — + which is why the sweep is nested rather than merged flat into extras.""" + u = extract_cloudflare_log( + { + "id": "log_2", + "cached": True, + "step": 3, + "tokens_in": 5, + "tokens_out": 1, + "provider": "anthropic", + "usage_metadata": {"cached": False, "step": 99, "log_id": "spoofed"}, + } + ) + assert u.extras["cached"] is True + assert u.extras["step"] == 3 + assert u.extras["log_id"] == "log_2" + assert u.extras["usage_metadata"] == {"cached": False, "step": 99, "log_id": "spoofed"} + + +def test_drift_omits_the_key_entirely_when_there_is_none() -> None: + """The common case must look exactly as it did before the sweep existed.""" + u = extract_cloudflare_log( + { + "id": "log_3", + "cached": False, + "step": 0, + "tokens_in": 9, + "tokens_out": 21, + "provider": "anthropic", + "usage_metadata": { + "input_tokens": 9, + "output_tokens": 21, + "total_tokens": 30, + "input_cached_tokens": 4, + }, + } + ) + assert u.extras == {"cached": False, "step": 0, "log_id": "log_3"} + assert "usage_metadata" not in u.extras + + +@pytest.mark.parametrize( + "key", + [ + "input_cached_tokens", + "inputCachedTokens", + "cachedContentTokenCount", + "cache_read_input_tokens", + "input_cache_creation_tokens", + "inputCacheCreationTokens", + "cache_creation_input_tokens", + "reasoningTokens", + "reasoning_tokens", + "thoughtsTokenCount", + "input_tokens", + "output_tokens", + "total_tokens", + ], +) +def test_drift_every_mapped_spelling_stays_out_of_the_sweep(key: str) -> None: + """A key that IS consumed must not also show up as drift — that would read as an + unhandled counter in reconciliation and invite double-counting.""" + u = extract_cloudflare_log( + { + "tokens_in": 100, + "tokens_out": 10, + "provider": "anthropic", + "usage_metadata": {key: 7}, + } + ) + assert "usage_metadata" not in u.extras, f"{key} should be mapped, not swept" diff --git a/tests/unit/test_queue.py b/tests/unit/test_queue.py index 0423f6b..9bd0b3c 100644 --- a/tests/unit/test_queue.py +++ b/tests/unit/test_queue.py @@ -273,7 +273,10 @@ def test_overflow_is_reported_through_on_error(): # aimed `max_batch_size` extra requests at a server that had just asked us to # slow down. # ---------------------------------------------------------------------- -@pytest.mark.parametrize("status", [413, 402, 415]) +# 402 is deliberately NOT here — see `test_402_is_held_not_dropped`. What makes a +# 413/415 batch fail is a property of the batch itself; a 402 is a property of the +# account and resolves out-of-band, so splitting it only drops every event faster. +@pytest.mark.parametrize("status", [413, 415]) def test_batch_only_4xx_is_split_not_head_of_line_blocked(status: int): """For these the SAME batch can never succeed, but its events can individually. @@ -303,7 +306,45 @@ def sender(batch): q.shutdown(timeout=1.0) -@pytest.mark.parametrize("status", [429, 408]) +def test_402_is_held_not_dropped(): + """A 402 must survive to be re-sent — it resolves out-of-band. + + Regression: 402 used to be permanent, which routed the batch to + `_send_individually`, where every isolated send 402ed too and was dropped for good. + Measured against a server returning 402: 5 events in, 6 HTTP calls out, 0 + recoverable — a lapsed Lago account silently discarded every billable event for the + whole outage. "Payment required" is a property of the ACCOUNT: it stops being true + the moment someone pays. + """ + attempts = {"n": 0} + delivered: list = [] + per_request_sizes: list = [] + + def sender(batch): + attempts["n"] += 1 + per_request_sizes.append(len(batch)) + # Account is lapsed for the first two attempts, then someone pays. + if attempts["n"] <= 2: + raise LagoApiError(402, '{"error":"payment required"}') + delivered.extend(batch) + + q = EventQueue(sender=sender, flush_interval=0.05, max_batch_size=10, max_retry_seconds=0.5) + try: + for i in "abcde": + q.push({"id": i}) + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline and not delivered: + time.sleep(0.05) + # Nothing lost: all five arrive once the account is current again. + assert [e["id"] for e in delivered] == list("abcde") + # Never fanned out — every request carried the whole batch, so no event was + # ever isolated and dropped. + assert all(n == 5 for n in per_request_sizes), per_request_sizes + finally: + q.shutdown(timeout=2.0) + + +@pytest.mark.parametrize("status", [429, 408, 402]) def test_throttling_4xx_is_retried_not_dropped(status: int): """A rate-limited or timed-out batch must reach Lago eventually. Dropping it loses revenue, and isolating it one-by-one multiplies the load on a @@ -331,7 +372,7 @@ def sender(batch): q.shutdown(timeout=2.0) -@pytest.mark.parametrize("status", [429, 408]) +@pytest.mark.parametrize("status", [429, 408, 402]) def test_throttling_4xx_applies_backoff(status: int): """The inverse of test_permanent_failure_does_not_apply_backoff: a throttling failure is transient, so it MUST leave a backoff in place —