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 —