diff --git a/CHANGELOG.md b/CHANGELOG.md index f405d1c..c633eb2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,7 +15,16 @@ All notable changes to this project will be documented here. Format follows [Kee ### Fixed +- **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. +- **`verify_ssl=False` could crash `LagoSDK()` construction.** The InsecureRequestWarning suppression reached through `requests.packages`, a legacy compatibility alias with no guarantee of existing, in an unguarded attribute chain inside `__init__`. Now `import urllib3` directly, wrapped: suppressing a warning must never fail construction. This sits on an advertised path — `verify_ssl` is a first-class constructor argument the docstring recommends for local dev — so the crash would have hit exactly the setup the flag was added to serve. +- **`WORKERS_AI_COMPAT_PREFIX` had drifted into two definitions.** `adapters/openai_native` decides the *provider* from it and `pricing` strips it before a catalog lookup; those two must never import each other, so it now lives in `canonical` (which imports nothing from the package — no cycle either way, and no pulling `pricing`'s ~50KB into a lightweight adapter). A drift between the copies would have been a silently unpriced call rather than a crash, so a test now asserts there is exactly one definition in the tree. +- **A moving `~` alias could overwrite a real listing's price, decided purely by catalog order.** Stripping OpenRouter's `~` marker indexes an alias under its real vendor, which is what makes a plain `-latest` id priceable at all — but it wrote the alias-derived keys with plain assignment. Collision-freedom was verified against the live catalog and still holds, so nothing is mispriced today; it was a property of that day's response rather than of the code. If OpenRouter ever lists both `google/gemini-flash-latest` and `~google/gemini-flash-latest`, whichever arrived later won: measured on a synthetic pair, the same lookup returned `0.009` or `0.001` depending only on position in the response. Alias-derived keys are now written only when absent, so a real listing always wins regardless of order; the `~`-spelled id still resolves to its own entry, and non-alias entries keep plain assignment so genuine duplicates behave exactly as before. +- **The `-\d{3}` version-strip arm is now scoped to OpenRouter, where it was the only thing that ever needed it.** Gemini's `model_version` can report a `-002` revision that OpenRouter omits from its ids, so the shared `_strip_version` grew a 3-digit arm — but that helper also builds the AWS/Bedrock price keys, and there a shortened key does not merely miss: `bedrock_model_key` feeds `table.setdefault(key, {})[direction] = price`, so two distinct models collapsing onto one key silently overwrite each other's rate. All four live catalogs are clean (OpenRouter 415 ids, Cloudflare 64, AWS offer 77, captured Bedrock 39 — zero model parts ending in exactly three digits), so this was latent, not active. Splitting it into an OpenRouter-only strip makes the risk structurally zero rather than empirically zero, and `amazon.titan-text-001` / `-002` now stay distinct keys. +- **`apply_markup`'s two bad-input fallbacks were not equivalent, and the ports disagreed on one.** An unparseable `usd` means the cost is unusable, so 0 is right; an unparseable `markup` means only the multiplier is unusable, and returning 0 there discards a perfectly good cost. Python returned `"0"` for both, JS fell back to 1.0 for a bad markup, so identical input would have produced different bills. Python now matches JS. This is defence in depth rather than a live fix: every `emit()` path already runs the customer's markup through `coerce_markup` (which falls back to 1.0 and reports under `on_error`), and `CostBreakdown.markup` / `fields[*]["cost"]` are `_fmt_money` output, so neither argument can actually arrive unparseable — the divergence was latent. What *was* untested is the end-to-end consequence of that guard, now pinned: a customer sending `markup="1,5"` gets the cost billed at 1.0 rather than zeroed, and the lost markup reaches `on_error`. - **A dropped event now always reaches `on_error`, not just a log line.** Two paths lost billable events while telling only the module logger, so a customer watching the error hook — the documented channel for every other billing gap — saw nothing: the queue's buffer **overflow**, and `emit()` dropping a call because no subscription resolved. The JS port already reported both, so the two repos disagreed on whether a lost event is even visible. + - **The overflow report is raised with the queue lock released, and is re-entrancy guarded per thread.** `_lock` is a plain `Lock`, not an `RLock`, so an `on_error` hook that itself emits — a plausible hook, since reporting a billing gap by emitting a metric is an obvious thing to do — deadlocked the *customer's* thread permanently if the report was made while holding it. Reporting after release fixes that but re-enters unboundedly, because the buffer is still full when the hook runs; a `threading.local` guard bounds it to one report per overflow per thread. Both failure modes are covered by a subprocess test, since an in-process deadlock wedges interpreter shutdown and hangs the whole suite rather than failing it. - **A negative token count could be emitted as a billable quantity.** `nonzero_numeric` filtered on truthiness, so `input=-100` survived and was pushed as `value="-100"`. Nothing upstream should produce one — every adapter clamps at extraction — but this is the last gate before an event is built, and the JS port already filtered on `> 0`. Now positive-only in both. - **`apply_markup` was the one money helper that could raise.** A non-numeric input hit a bare `Decimal("abc")` and threw `InvalidOperation` from inside `_push_cost_event`, under `emit()`'s catch-all — so the cost event was dropped and reported as an unknown `"emit"` error instead of taking the documented no-price path, past every caller relying on this module's `None`-on-bad-input convention. Now parsed through `_parse_price` and floored to `0`, matching the JS port and `compute_precomputed_cost`'s existing floor-to-zero behaviour. `money_golden.json` is unchanged and still byte-identical across repos. @@ -42,6 +51,7 @@ All notable changes to this project will be documented here. Format follows [Kee - Worth recording an under-report the review didn't name: for an **additive** provider the old basis was far wider than the reasoning case. An Anthropic call with `input=1000`, `cache_read=900`, `output=100` reported `unit=1100` against 2000 actually consumed — cache tokens are real extra consumption there, not a subset. Verified end-to-end against a live Lago instance: the `llm_cost` charge's `units` moved by exactly 2000, with the dynamic charge carrying the metered `$0.05`. - **A config-only `api_url` sent every event to PRODUCTION Lago.** `LagoSDK.__init__` defaulted `api_url` to the production URL, so the `if api_url:` guard meant to let explicit args win *always* fired — silently overwriting whatever the caller had set on `LagoConfig`. `LagoSDK(api_key=k, config=LagoConfig(api_url="http://localhost:3000/api/v1"))` therefore shipped a local-dev customer's usage data to `api.getlago.com`, with no error and nothing in the logs. The default is now `None`, and every explicit arg is guarded on "was it actually passed?" rather than on truthiness, so a config value survives when the arg was omitted while an explicit arg still wins when given. The production default is unchanged when nothing is passed at all. + - **`api_url` is the one field guarded on non-emptiness rather than on presence.** `api_url=os.environ.get("LAGO_API_URL", "")` with the variable unset — the ordinary way to write this — would otherwise store `""`, and an empty base URL is unrecoverable downstream: `requests` raises `MissingSchema`, which is not a `LagoApiError`, so the queue classifies it as transient and retries at the 60s ceiling forever. Every event is buffered and none is ever delivered, with a growing buffer as the only symptom. An empty string now falls through to config, then to the production default, exactly as an omitted argument does. - **This was the shortest path to the bug, not an exotic one.** A custom `api_url` and `verify_ssl=False` go together in exactly one setup — a local dev Lago behind a self-signed cert (Traefik's default) — which is the setup `verify_ssl` was added for. Since `verify_ssl` was reachable *only* through a `LagoConfig`, the feature pushed callers straight into the clobber. `LagoSDK(..., verify_ssl=False)` is now accepted directly, so that setup needs no config object at all. - **The JS port had the same bug inverted, so the two repos disagreed.** `...(opts.config || {})` was spread LAST, so a `config.apiUrl` overrode an *explicitly passed* `apiUrl` — the opposite of what Python documents and does. The identical call therefore billed a different Lago instance depending on which SDK you used. Both now apply explicit options over config, and `verifySsl` is accepted at the top level to match. @@ -53,7 +63,8 @@ All notable changes to this project will be documented here. Format follows [Kee - **A backfill re-run after the price table warmed up silently discarded the priced events.** `_emit_token_events` and the per-field branch of `_push_cost_event` both built `f"{event_id}_{field_name}"` over the same field vocabulary (`input`, `output`, `cache_read`, ...), and both are reachable for the SAME `event_id`: `emit()` falls back to token events when a price lookup misses, then takes the cost path once the table is warm. So run 1 (cold table) sent `backfill_X_input` under `llm_input_tokens`, and run 2 (warm) re-sent the identical id under `llm_cost` — Lago rejected it as a duplicate `transaction_id`, and because `/events/batch` is all-or-nothing that rejection failed **every other event in the batch** too. Net effect: the dollar amounts for that window were never billed, only the raw token counts, and nothing surfaced it — which defeats the very idempotency promise `event_id` exists to provide. The two multi-event paths now use disjoint namespaces (`_tok_` / `_cost_`); the single precomputed-cost event stays unsuffixed, since one event has nothing to disambiguate. No migration needed: `event_id` is absent from 0.2.0 entirely, so no released path has ever emitted the old format (they all use a random UUID). Pinned by a regression test that asserts a cold run's ids and a warm run's ids over the same `event_id` do not intersect — verified to fail without the fix, where all four ids collided. -- **Rate-limited events were dropped for good, and the retry made the rate limit worse.** `_is_permanent_failure` classified the whole `400-499` range as unretryable, but two 4xx statuses mean "try again, later": **429** (rate limited) and **408** (request timeout). So a throttled 100-event batch took the isolate-and-drop path — 100 further requests aimed at the server that had just asked us to slow down, each of them also throttled, each then logged and discarded. 100 billable events lost, and the isolation actively deepened the throttle. This was a regression: before the permanent/transient split existed, the same 429 went through the 1s→60s backoff and eventually landed. "Permanent" is now an explicit set (`400, 401, 403, 404, 409, 422`) rather than a range, so an unrecognized 4xx errs toward retrying — waiting on an event that would have been dropped costs latency, dropping one that would have been accepted costs revenue. The isolate-one-by-one recovery is unchanged for the validation statuses it was written for, so a single duplicate `transaction_id` still doesn't take its batch down with it. +- **Rate-limited events were dropped for good, and the retry made the rate limit worse.** `_is_permanent_failure` classified the whole `400-499` range as unretryable, but two 4xx statuses mean "try again, later": **429** (rate limited) and **408** (request timeout). So a throttled 100-event batch took the isolate-and-drop path — 100 further requests aimed at the server that had just asked us to slow down, each of them also throttled, each then logged and discarded. 100 billable events lost, and the isolation actively deepened the throttle. This was a regression: before the permanent/transient split existed, the same 429 went through the 1s→60s backoff and eventually landed. "Permanent" is now an explicit set (`400, 401, 402, 403, 404, 409, 413, 415, 422`) rather than a range, so an unrecognized 4xx errs toward retrying — waiting on an event that would have been dropped costs latency, dropping one that would have been accepted costs revenue. The isolate-one-by-one recovery is unchanged for the validation statuses it was written for, so a single duplicate `transaction_id` still doesn't take its batch down with it. + - **Three of those statuses are permanent for the opposite reason to 429, and that is why they belong in the set.** `413` (payload too large), `402` (payment required) and `415` (unsupported media type) reject the batch *as a batch* — re-sending the identical bytes can never succeed, so retrying it re-prepends it at the head of the FIFO and blocks every event behind it until the 60s ceiling, indefinitely. Classifying them permanent routes them to the isolate-one-by-one path, which splits the batch and delivers what is individually deliverable — the case that path was built for, and the one it could not previously reach. `405` and `410` stay transient: both are plausibly a misconfigured or mid-deploy endpoint that will start working again. - **Every documented way to call Workers AI went unpriced.** `_infer_provider` matched only a bare `@cf/...`, but Cloudflare's OpenAI-compatible `/compat` endpoint requires the routing-prefixed form — `workers-ai/@cf/meta/llama-3.3-70b-instruct-fp8-fast` — which is exactly what `README` and `examples/cloudflare_gateway_demo.ipynb` prescribe, and the only form a **streaming** call can report (the synthetic usage payload carries no model, so `resolve_model` returns the requested string verbatim). Those calls were stamped `provider="openai"`, looked up against OpenRouter as `openai/workers-ai/@cf/...`, missed, and silently degraded to token events — with nothing to degrade *to* in an `llm_cost`-only billing setup. The `workers-ai` entry in `_INPUT_INCLUDES_CACHE_READ` never applied either, so on the paths that did price, cache reads were billed twice again. Fixing the provider check alone was not enough: Cloudflare's catalog keys models as bare `@cf/...`, so `lookup_cloudflare_workers_ai` now strips the routing prefix before matching. `CanonicalUsage.model` deliberately keeps the spelling the customer used, so reporting stays faithful to the request while pricing resolves. @@ -62,6 +73,7 @@ All notable changes to this project will be documented here. Format follows [Kee - **Gateway-backfilled Gemini calls could never be priced, and their cached tokens were billed twice.** `extract_cloudflare_log` passed Cloudflare's own provider vocabulary through verbatim, but that is not the vocabulary the pricing and token-semantics tables key off — and not even Cloudflare's own URL slug (the logs say `workers-ai` where the endpoint path says `workersai`). A real captured entry reports `provider: "google-ai-studio"`, which matched no vendor in `_VENDOR_MAP`, so `lookup_openrouter` searched a vendor that does not exist and missed every time (verified against the live 400-model OpenRouter table: miss as `google-ai-studio`, hit as `gemini`). The same miss also kept it out of `_INPUT_INCLUDES_CACHE_READ`, so Gemini's `cache_read` — a **subset** of its input count — was billed on top of the full input rather than subtracted from it. Cloudflare's names are now mapped onto the SDK's (`google-ai-studio`/`google-vertex-ai`/`vertex` → `gemini`, `azure-openai`/`azureopenai` → `openai`, `workersai` → `workers-ai`); anything unrecognized passes through untouched, since a clean miss falling back to token events beats an invented mapping. AWS Bedrock is deliberately **not** mapped — its prices key off `api.startswith("bedrock")` and this connector always sets `api="cloudflare_gateway"`, so a mapping would route it to OpenRouter under a vendor that cannot match. - **A model already carrying its vendor prefix never matched a price.** A real gateway log for a REST-path call reports `model: "anthropic/claude-opus-4.8"` alongside `provider: "anthropic"`, which `lookup_openrouter` turned into `"anthropic/anthropic/claude-opus-4.8"` — a guaranteed miss. The prefix is now stripped, but **only** when it agrees with the vendor resolved from `provider`, so the lookup stays vendor-gated as documented: a model naming a different vendor than the call claims is still a miss, not a cross-vendor mispricing. With both fixes, all 10 distinct (provider, model) pairs across the real captured fixtures now resolve to a live price; three of them previously missed. - **Workers AI cached tokens were billed twice.** `provider="workers-ai"` was missing from `_INPUT_INCLUDES_CACHE_READ`, but Workers AI is only ever reached through Cloudflare's OpenAI-**compatible** endpoint (`.../compat`), so its usage payload is the OpenAI shape — `prompt_tokens` already **includes** `prompt_tokens_details.cached_tokens`. It is a distinct provider only because it prices against Cloudflare's own catalog, not because its token semantics differ. With the cached portion never subtracted from `input`, those tokens were charged once at the full input rate *and* again at the cache-read rate, which Cloudflare's catalog does publish (verified live: `@cf/moonshotai/kimi-k2.6`, `@cf/moonshotai/kimi-k2.7-code` and `@cf/zai-org/glm-5.2` all list a "per M cached input tokens" price). Measured against a real cached call (prompt 23233 / cached 23168) at live catalog rates, this overbilled by **+583%**; the error scales with cache hit rate, so a long cached system prompt — the standard agent workload — is the worst case. Pinned by two new golden cases (`workers-ai` subtracts, `anthropic` stays additive) carrying those real counts. + - **The same omission existed on the output side and is fixed with it.** `workers-ai` was missing from `_OUTPUT_INCLUDES_REASONING` too, and for the identical reason: the `/compat` endpoint returns the OpenAI shape, where `completion_tokens_details.reasoning_tokens` is a **subset** of `completion_tokens`, so counting it additively inflates the token basis — a 100-input/1000-output call with 800 reasoning tokens reported 1900 instead of 1100. **Unlike the cache-read side, this one is not reachable on a live call today, and the 1900 above is the arithmetic, not a measured bill.** Verified against real Workers AI on 2026-08-20: three reasoning-capable models (`@cf/openai/gpt-oss-120b`, `@cf/deepseek-ai/deepseek-r1-distill-qwen-32b`, `@cf/qwen/qwen3-30b-a3b-fp8`) returned no `completion_tokens_details` block at all, and the gateway Logs API reports only `input_tokens`/`output_tokens`/`total_tokens`/`input_cached_tokens`/`neurons`/`units` on `workers-ai` rows — so `reasoning` is never populated for this provider by either the native `/compat` adapter or the logs backfill. It is fixed regardless, because the two sets must agree about the same provider for the same reason: the cache-read side *is* live-reachable (`input_cached_tokens` is reported and non-null), and one set knowing that `/compat` implies OpenAI token semantics while the other does not is precisely the asymmetry that produced the original double-billing. - **`_parse_price` raised instead of returning `None` on absurdly large values.** `.quantize()` sat outside the `try`, so any value ≥ 1e16 (16 integer + 12 fractional digits exceeds `Decimal`'s default 28-digit context precision) threw `InvalidOperation` straight out of a function documented as returning `None` on bad input — past every caller relying on that, and out of `compute_precomputed_cost` into `emit()`'s catch-all, where the event was dropped as an unknown error rather than taking the normal "no price" path. Now returns `None`, which also matches what the JS port returns for the same inputs. - **Cross-repo golden fixture gained a `precomputed_cases` section**, asserted by both repos, carrying verbatim `cost` values from real Cloudflare AI Gateway log entries — including ones below 1e-6, where JS renders the number in exponential notation. Python has always parsed those correctly; the JS port silently billed them as $0 (fixed in that repo's matching release). `cases` also gained an optional `provider`, so per-provider token semantics are now pinned by the shared fixture rather than by each repo's own tests. diff --git a/src/lago_agent_sdk/adapters/openai_native.py b/src/lago_agent_sdk/adapters/openai_native.py index 5317b8f..b10fee6 100644 --- a/src/lago_agent_sdk/adapters/openai_native.py +++ b/src/lago_agent_sdk/adapters/openai_native.py @@ -39,7 +39,7 @@ from typing import Any, cast -from ..canonical import CanonicalUsage +from ..canonical import WORKERS_AI_COMPAT_PREFIX, CanonicalUsage from ._common import resolve_model # Cloudflare Workers AI names every model "@cf//". Reaching one @@ -49,7 +49,6 @@ # strips the routing prefix before matching, because Cloudflare's own catalog lists # only the bare form. _WORKERS_AI_MODEL_PREFIX = "@cf/" -_WORKERS_AI_COMPAT_PREFIX = "workers-ai/" # Top-level usage fields we recognize across BOTH chat completions and responses APIs. _KNOWN_USAGE_FIELDS = { @@ -131,7 +130,7 @@ def _infer_provider(resolved_model: str) -> str: priced against OpenRouter, missed, and silently degraded to token events. """ if resolved_model.startswith(_WORKERS_AI_MODEL_PREFIX) or resolved_model.startswith( - f"{_WORKERS_AI_COMPAT_PREFIX}{_WORKERS_AI_MODEL_PREFIX}" + f"{WORKERS_AI_COMPAT_PREFIX}{_WORKERS_AI_MODEL_PREFIX}" ): return "workers-ai" return "openai" diff --git a/src/lago_agent_sdk/canonical.py b/src/lago_agent_sdk/canonical.py index a47f079..054aa22 100644 --- a/src/lago_agent_sdk/canonical.py +++ b/src/lago_agent_sdk/canonical.py @@ -5,6 +5,21 @@ from dataclasses import asdict, dataclass, field from typing import Any +# The routing prefix Cloudflare's OpenAI-compatible `/compat` endpoint requires: +# "workers-ai/@cf/meta/llama-3.3-70b-instruct-fp8-fast". The same model therefore +# arrives under two spellings depending on which surface the customer used, and two +# unrelated layers need to agree on this string — `adapters/openai_native` decides +# the PROVIDER from it, and `pricing.lookup_cloudflare_workers_ai` strips it before +# matching, because Cloudflare's own catalog lists only the bare "@cf/..." form. +# +# It lives here rather than in either of them because they must never import each +# other (an adapter is a pure function of a provider response; pricing is SDK state), +# and because a drift between two copies is a silent unpriced call, not a crash. This +# module is the natural shared floor: it imports nothing from the package, so there is +# no cycle in either direction, and depending on it does not pull `pricing`'s ~50KB +# into a lightweight adapter. +WORKERS_AI_COMPAT_PREFIX = "workers-ai/" + @dataclass class CanonicalUsage: @@ -49,5 +64,19 @@ def nonzero_numeric(self) -> dict[str, int]: """ return {k: v for k in self.NUMERIC_FIELDS if (v := getattr(self, k)) and v > 0} + def negative_numeric(self) -> dict[str, int]: + """Fields `nonzero_numeric` DROPPED for being negative, so the caller can + report them. + + Reachable, unlike most defensive paths here: `CanonicalUsage` is exported and + `emit()` takes one directly, which is the documented way to backfill usage the + SDK did not intercept. A caller computing a delta wrongly can hand us a + negative, and silently dropping it is the one drop path that never reached + `on_error` — the same gap that was closed for queue overflow and for an + unresolvable subscription. Kept as a separate pure query so `CanonicalUsage` + stays a dumb dataclass with no notification channel of its own. + """ + return {k: v for k in self.NUMERIC_FIELDS if (v := getattr(self, k)) and v < 0} + def to_dict(self) -> dict[str, Any]: return asdict(self) diff --git a/src/lago_agent_sdk/lago_client.py b/src/lago_agent_sdk/lago_client.py index 55ff810..01f03f4 100644 --- a/src/lago_agent_sdk/lago_client.py +++ b/src/lago_agent_sdk/lago_client.py @@ -20,11 +20,20 @@ def __init__(self, api_key: str, api_url: str, timeout: float = 10.0, verify_ssl # The customer explicitly opted out via config — they've already # accepted the risk; requests/urllib3's warning on every single # request would just be noise at that point, not new information. - # Access urllib3 via requests' own re-export — it's only a - # transitive dependency for us, not one we declare directly. - requests.packages.urllib3.disable_warnings( # type: ignore[attr-defined] - requests.packages.urllib3.exceptions.InsecureRequestWarning # type: ignore[attr-defined] - ) + # Import urllib3 directly rather than through `requests.packages`, + # which is a legacy compatibility shim that is not guaranteed to exist. + # Wrapped because this is an optional convenience: suppressing a warning + # must never be able to fail construction of the SDK itself. That is not + # hypothetical — `verify_ssl=False` is now a first-class constructor + # argument that the docstring recommends for local dev, so this line sits + # on an advertised path, and an ImportError/AttributeError here would + # take down `LagoSDK()` for the exact setup the flag was added to serve. + try: + import urllib3 + + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + except Exception: # noqa: BLE001 + pass def __repr__(self) -> str: if not self.api_key: diff --git a/src/lago_agent_sdk/pricing.py b/src/lago_agent_sdk/pricing.py index ad2ff40..7475d8f 100644 --- a/src/lago_agent_sdk/pricing.py +++ b/src/lago_agent_sdk/pricing.py @@ -51,7 +51,7 @@ from decimal import ROUND_DOWN, Decimal, InvalidOperation from typing import Any, Protocol -from .canonical import CanonicalUsage +from .canonical import WORKERS_AI_COMPAT_PREFIX, CanonicalUsage logger = logging.getLogger("lago_agent_sdk.pricing") @@ -84,7 +84,17 @@ # tokens (reasoning is a subset of output). For these, reasoning is billed as # part of output and must NOT be billed again separately. (Gemini's `thoughts` # are additive to output, so it's absent here.) -_OUTPUT_INCLUDES_REASONING = frozenset({"openai"}) +# "workers-ai" belongs here for the same reason it is in _INPUT_INCLUDES_CACHE_READ +# above: it is only ever reached through Cloudflare's OpenAI-COMPATIBLE endpoint, so +# its usage payload is the OpenAI shape — and in that shape +# `completion_tokens_details.reasoning_tokens` is a SUBSET of `completion_tokens`, +# exactly as it is for real OpenAI. `extract_openai_native` fills `reasoning` from that +# key with no provider gate, so omitting it counted the subset twice: measured, a +# 100/1000/reasoning-800 call reported unit=1900 against 1100 consumed. `compute_cost` +# would double-BILL the same tokens and does not today only because +# _CLOUDFLARE_UNIT_FIELD_MAP happens to carry no reasoning unit — an accident, not a +# guard, and Cloudflare hosts reasoning models (deepseek-r1, qwen, glm). +_OUTPUT_INCLUDES_REASONING = frozenset({"openai", "workers-ai"}) # Canonical field -> OpenRouter pricing key. _OPENROUTER_FIELD_MAP = { @@ -117,12 +127,6 @@ "per M cached input tokens": "cache_read", } -# The routing prefix the gateway's OpenAI-compatible `/compat` endpoint requires. -# Cloudflare's catalog keys models as bare "@cf/...", so this comes off before a -# lookup. Kept in sync with `adapters/openai_native._WORKERS_AI_COMPAT_PREFIX`, -# which decides the provider from the same two spellings. -_WORKERS_AI_COMPAT_PREFIX = "workers-ai/" - # Cloudflare's catalog page size, and a hard bound on the paging loop. The loop runs # on the queue's flush tick ahead of the drain, so it must terminate even if the # endpoint keeps returning full pages. 40 pages covers ~2000 models against a real @@ -154,12 +158,21 @@ _SCALE = 12 _Q = Decimal(1).scaleb(-_SCALE) # Decimal("1E-12") # A trailing version/revision marker OpenRouter usually omits from its own ids. -# Shapes seen live: Anthropic's compact date ("-20250929"), an explicit "-v2", and +# Shapes seen live: Anthropic's compact date ("-20250929") and an explicit "-v2". +_VERSION_DATE_SUFFIX = re.compile(r"-(?:\d{8}|v\d+)$") + # Gemini's 3-digit revision ("-002", which `model_version` can report where -# OpenRouter lists only the bare name). Verified safe against the live 415-model -# catalog: ZERO ids have a model part ending in exactly three digits, so the -# "-\d{3}" arm cannot shorten a real listing. -_VERSION_DATE_SUFFIX = re.compile(r"-(?:\d{8}|\d{3}|v\d+)$") +# OpenRouter lists only the bare name) is stripped for OpenRouter matching ONLY. +# It is deliberately NOT in the shared `_strip_version`: that helper also builds +# the AWS/Bedrock price keys, where a shortened key does not merely miss but +# silently MIS-prices — `bedrock_model_key` feeds +# `table.setdefault(key, {})[direction] = price`, so two distinct models +# collapsing to one key overwrite each other's rate. All four live catalogs are +# currently clean (OpenRouter 415 ids, Cloudflare 64, AWS offer 77, captured +# Bedrock 39: zero model parts end in exactly three digits), but the arm was only +# ever motivated by OpenRouter, and scoping it makes that risk structurally zero +# instead of empirically zero. +_OPENROUTER_VERSION_SUFFIX = re.compile(r"-(?:\d{8}|\d{3}|v\d+)$") # ---------------------------------------------------------------------- @@ -207,13 +220,28 @@ def apply_markup(usd: str, markup: str) -> str: dropped and reported as an unknown "emit" error instead of taking the documented no-price path. It was also the one money helper in this module that could raise at all, past every caller relying on the `None`-on-bad-input - convention. The JS port already defaulted to zero here, so the two repos were - differently wrong on the same input. + convention. + + Both fallbacks are DEFENCE IN DEPTH, not live behaviour: every `emit()` path + runs the customer's markup through `coerce_markup` first (which falls back to + 1.0 and reports under "pricing"), and `CostBreakdown.markup` / + `fields[*]["cost"]` are `_fmt_money` output, so neither argument can actually + arrive unparseable here. They are still not interchangeable, and the two ports + disagreed on them: + + - An unparseable `usd` means the cost itself is unusable — nothing to bill: 0. + - An unparseable `markup` means only the MULTIPLIER is unusable. Returning 0 + there would discard a good cost, an under-bill to nothing; 1.0 bills the real + cost with no markup, the smallest defensible error. JS already did this; + Python returned "0", so identical input produced different bills if anything + ever did reach it. Aligned rather than left as a latent divergence. """ base = _parse_price(usd) mult = _parse_price(markup) - if base is None or mult is None: + if base is None: return _fmt_money(Decimal(0)) + if mult is None: + mult = Decimal(1) return _fmt_money((base * mult).quantize(_Q, rounding=ROUND_DOWN)) @@ -241,6 +269,12 @@ def _strip_version(model: str) -> str: return _VERSION_DATE_SUFFIX.sub("", model) +def _strip_version_openrouter(model: str) -> str: + """`_strip_version`, plus Gemini's 3-digit revision. OpenRouter matching only — + see `_OPENROUTER_VERSION_SUFFIX` for why this is not the shared helper.""" + return _OPENROUTER_VERSION_SUFFIX.sub("", model) + + # ---------------------------------------------------------------------- # Price tables # ---------------------------------------------------------------------- @@ -419,13 +453,28 @@ def parse_openrouter(data: Any) -> dict[str, Any]: # llm_cost-only setup. Stripping the marker indexes them under their real # vendor. Verified collision-free against the live catalog: no un-prefixed # id duplicates a "~"-prefixed one, so nothing is overwritten. + # + # `setdefault` for the alias-derived keys rather than assignment: the + # collision-freedom above is a property of TODAY's catalog, and with plain + # assignment the winner depended purely on iteration order — if OpenRouter + # ever ships both "google/gemini-flash-latest" and + # "~google/gemini-flash-latest", the moving alias could overwrite the real + # listing's rate (measured on a synthetic pair: 0.009 vs 0.001 for the same + # lookup, decided by nothing but position in the response). A real listing + # now always wins, whatever the order. Non-alias entries keep plain + # assignment so genuine duplicates behave exactly as before. bare = mid[1:] if mid.startswith("~") else mid + is_alias = bare != mid exact[mid] = mp - if bare != mid: - exact[bare] = mp + if is_alias: + exact.setdefault(bare, mp) if "/" in bare: vendor, _, suffix = bare.partition("/") - norm[(vendor.lower(), _norm(suffix))] = mp + norm_key = (vendor.lower(), _norm(suffix)) + if is_alias: + norm.setdefault(norm_key, mp) + else: + norm[norm_key] = mp return {"exact": exact, "norm": norm} @@ -574,7 +623,7 @@ def lookup_openrouter(table: dict[str, Any], provider: str, model: str) -> Model if hit is not None: return hit # 3. date/version-stripped, normalized - hit = norm.get((vendor, _norm(_strip_version(model)))) + hit = norm.get((vendor, _norm(_strip_version_openrouter(model)))) if hit is not None: return hit return None @@ -654,7 +703,7 @@ def lookup_cloudflare_workers_ai(table: dict[str, ModelPrice], model: str) -> Mo the only form a streaming call can report. Without the strip, recognising the prefixed spelling as Workers AI upstream just moves the miss here. """ - for candidate in (model, model.removeprefix(_WORKERS_AI_COMPAT_PREFIX)): + for candidate in (model, model.removeprefix(WORKERS_AI_COMPAT_PREFIX)): hit = table.get(candidate) if hit is not None: return hit diff --git a/src/lago_agent_sdk/queue.py b/src/lago_agent_sdk/queue.py index 5e1495c..18c59d0 100644 --- a/src/lago_agent_sdk/queue.py +++ b/src/lago_agent_sdk/queue.py @@ -38,7 +38,17 @@ # (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. -_PERMANENT_STATUSES = frozenset({400, 401, 403, 404, 409, 422}) +# +# 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}) def _is_permanent_failure(exc: Exception) -> bool: @@ -85,6 +95,8 @@ def __init__( self._stopping = threading.Event() self._backoff_seconds = 0.0 self._http_calls = 0 # for tests + # Per-thread "already reporting an overflow" flag — see push(). + self._reporting = threading.local() self._thread = threading.Thread(target=self._run, name="lago-queue", daemon=True) self._thread.start() @@ -104,6 +116,7 @@ def _after_in_child(self) -> None: self._buffer = deque() # don't replay parent's events from the child self._backoff_seconds = 0.0 self._http_calls = 0 + self._reporting = threading.local() # Note: the PricingProvider self-heals on fork via a PID check inside # lookup()/maybe_refresh(); we deliberately do NOT call into it from this # fork handler (touching it here changes thread timing enough to trip @@ -121,22 +134,45 @@ def wake(self) -> None: def push(self, event: dict[str, Any]) -> None: with self._lock: - if len(self._buffer) >= self._max_buffer_size: + overflowed = len(self._buffer) >= self._max_buffer_size + if overflowed: self._buffer.popleft() + self._buffer.append(event) + should_wake = len(self._buffer) >= self._max_batch_size + # Both of these run with the lock RELEASED, the same shape `should_wake` + # already used. Reporting inside the lock deadlocked the caller: `_lock` is a + # plain Lock, not an RLock, so a customer `on_error` that touched the SDK at + # all — emitting a diagnostic, forcing a flush — blocked forever on a lock its + # own thread already held. Overflow happens under sustained load, which is + # exactly when such a hook fires, and the failure is worse than the drop it + # reports: an unnoticed dropped event costs one event, a hung producer thread + # costs the application. The surrounding try/except cannot help, because a + # deadlock is not an exception. + # + # Keeping the callback out of the lock also stops a full buffer from running + # the hook plus a log write synchronously on the customer's LLM-call thread + # while holding the lock every producer and the drain thread need. + if overflowed and not getattr(self._reporting, "active", False): + # Re-entrancy guard, per thread. Moving the report out of the lock fixed + # the deadlock but exposed the other half: the buffer is full again by the + # time the hook runs, so a hook that calls `push()` overflows again and + # re-enters without bound. Suppressing the nested report breaks the cycle + # while still letting the hook's own event be buffered. Per-thread so one + # producer's hook can never silence another producer's report. + self._reporting.active = True + try: logger.warning("lago queue overflow at %d events; dropping oldest", self._max_buffer_size) # Also through on_error: an overflow drops BILLABLE events, and a # customer watching only the error hook — the documented channel for - # every other billing gap — never learned revenue had been lost. The - # JS port already reported this, so the two disagreed on whether a - # dropped event is visible. + # every other billing gap — never learned revenue had been lost. self._report_error( RuntimeError( f"queue overflow at {self._max_buffer_size} events; dropped the oldest event" ), "overflow", ) - self._buffer.append(event) - should_wake = len(self._buffer) >= self._max_batch_size + finally: + self._reporting.active = False if should_wake: self._wake.set() @@ -196,6 +232,13 @@ def _send_individually(self, batch: list[dict[str, Any]], batch_exc: Exception) for what's really one root cause. """ self._report_error(batch_exc) + # Collected and re-queued ONCE at the end, not per event. `_replay_failed` + # prepends, so calling it inside the loop reversed the survivors' relative + # order: a 413 batch of a,b,c,d,e whose b,c,d fail transiently 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 + # meaningful — so a recovery path must not silently invert it. + retry: list[dict[str, Any]] = [] for event in batch: try: self._http_calls += 1 @@ -209,7 +252,9 @@ def _send_individually(self, batch: list[dict[str, Any]], batch_exc: Exception) ) else: logger.warning("lago send failed for isolated event, will retry: %s", exc) - self._replay_failed([event]) + retry.append(event) + if retry: + self._replay_failed(retry) def _run(self) -> None: while not self._stopping.is_set(): diff --git a/src/lago_agent_sdk/sdk.py b/src/lago_agent_sdk/sdk.py index df4ee31..c9edfbd 100644 --- a/src/lago_agent_sdk/sdk.py +++ b/src/lago_agent_sdk/sdk.py @@ -58,10 +58,19 @@ def __init__( reachable with ``LagoSDK(api_key=..., api_url=..., verify_ssl=False)``. """ self.config = config or LagoConfig(api_key=api_key) - # explicit args win over `config` — each guarded on "was it actually - # passed?", never on truthiness, so a config value survives when it wasn't. + # explicit args win over `config` — guarded on "was it actually passed?" + # rather than on truthiness, so a config value survives when it wasn't. self.config.api_key = api_key or self.config.api_key - if api_url is not None: + # `api_url` is the one exception: an EMPTY string must not win either. The bug + # this guard was written for was a *truthy default* overwriting config, so + # accepting "" swapped one silent misroute for a worse one — + # `api_url=os.environ.get("LAGO_API_URL", "")` with the var unset used to keep + # the production URL and instead wrote "". Downstream that is unrecoverable: + # `requests` raises MissingSchema, which is not a LagoApiError, so the queue + # 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. + if api_url: self.config.api_url = api_url if default_subscription_id is not None: self.config.default_subscription_id = default_subscription_id @@ -269,13 +278,11 @@ def emit( try: sub = self._resolve_subscription(subscription) if not sub: - # Reported as well as logged: this drops the call's billing entirely, - # and a customer watching on_error — the documented channel for every - # other billing gap — saw nothing. The JS port already reported it. - logger.error( - "lago: dropping events for model=%s — no resolvable subscription", - usage.model, - ) + # `_report_error` is the single channel: it invokes on_error AND + # logs. An extra logger.error here emitted the same drop twice under + # two different levels, so a customer grepping logs counted one lost + # call as two — and the JS port logged nothing at all, so the two + # repos reported 2 lines vs 0 for the same event. self._report_error( ValueError( f"no resolvable subscription for model={usage.model!r}; events dropped. " @@ -339,6 +346,18 @@ def _emit_token_events( self, usage: CanonicalUsage, sub: str, dimensions: dict[str, Any] | None, event_id: str | None = None ) -> None: nonzero = usage.nonzero_numeric() + # A negative count is silently unbillable — Lago would otherwise sum it into + # a negative quantity. It was the only drop path in the SDK that never + # reached on_error, so a caller who built a CanonicalUsage with a bad delta + # saw nothing at all. Reported before the empty-check, because an event whose + # only fields were negative leaves `nonzero` empty and would return below + # without a word. + negatives = usage.negative_numeric() + if negatives: + self._report_error( + ValueError(f"dropped negative token counts for model={usage.model!r}: {negatives}"), + "negative_tokens", + ) if not nonzero: # Mistral legacy / empty — nothing to bill return diff --git a/tests/unit/test_buffer_overflow.py b/tests/unit/test_buffer_overflow.py index d1c9907..a791393 100644 --- a/tests/unit/test_buffer_overflow.py +++ b/tests/unit/test_buffer_overflow.py @@ -14,10 +14,18 @@ def test_overflow_drops_oldest_at_exact_boundary(): def slow_sender(batch): paused.wait(timeout=30.0) + # max_batch_size > max_buffer_size keeps the background worker from ever being + # woken by push (the buffer can't reach max_batch_size), the same determinism + # fix test_repeated_overflow_keeps_window_sliding already carries. Without it + # push() signals the worker on the 10_000th event and the worker then races the + # assertions below for `_lock` — it is entitled to drain at any point once woken, + # so "the buffer still holds what I just pushed" is not a property the test can + # rely on. Nothing here needs the worker to run; every assertion is about buffer + # contents, and the finally block only unpauses so shutdown can finish. q = EventQueue( sender=slow_sender, flush_interval=10.0, # never timer-flush during the test - max_batch_size=10_000, # match buffer so worker takes everything once unpaused + max_batch_size=20_000, max_buffer_size=10_000, ) try: diff --git a/tests/unit/test_pricing.py b/tests/unit/test_pricing.py index 85955ea..c66079b 100644 --- a/tests/unit/test_pricing.py +++ b/tests/unit/test_pricing.py @@ -4,6 +4,7 @@ import json import pathlib +import re import uuid from decimal import Decimal from typing import Any @@ -12,11 +13,13 @@ from lago_agent_sdk import CanonicalUsage, LagoConfig, LagoSDK, ModelPrice from lago_agent_sdk.adapters.openai_native import extract_openai_native +from lago_agent_sdk.canonical import WORKERS_AI_COMPAT_PREFIX from lago_agent_sdk.pricing import ( HttpPricingFetcher, PricingProvider, _parse_price, _pick_mistral_canonical, + apply_markup, bedrock_model_key, coerce_markup, compute_cost, @@ -243,6 +246,60 @@ def test_openrouter_date_version_stripped_match() -> None: assert mp.input == Decimal("0.000001") +@pytest.mark.parametrize("alias_first", [True, False]) +def test_openrouter_moving_alias_never_overwrites_a_real_listing(alias_first: bool) -> None: + """A "~" alias and a real listing can collide on the same key. Which one wins + must not depend on catalog order — with plain assignment it did, and the moving + alias's rate (0.009) could replace the real listing's (0.001) purely by + arriving later in the response.""" + real = {"id": "google/gemini-flash-latest", "pricing": {"prompt": "0.001"}} + alias = {"id": "~google/gemini-flash-latest", "pricing": {"prompt": "0.009"}} + data = [alias, real] if alias_first else [real, alias] + table = parse_openrouter({"data": data}) + mp = lookup_openrouter(table, "gemini", "gemini-flash-latest") + assert mp is not None + assert mp.input == Decimal("0.001"), ( + f"real listing must win regardless of order (alias_first={alias_first})" + ) + # the "~"-spelled id still resolves to its own entry + assert table["exact"]["~google/gemini-flash-latest"].input == Decimal("0.009") + + +def test_three_digit_revision_is_stripped_for_openrouter_only() -> None: + """The "-002" arm exists for Gemini's revision, which only OpenRouter omits. + + It must NOT reach the AWS/Bedrock key builder: there a shortened key does not + merely miss, it collapses two models onto one key whose per-direction prices + are assigned in place, so one silently overwrites the other's rate. + """ + table = parse_openrouter({"data": [{"id": "google/gemini-2.5-flash", "pricing": {"prompt": "0.001"}}]}) + assert lookup_openrouter(table, "gemini", "gemini-2.5-flash-002") is not None + + # shared helper keeps a 3-digit tail, so distinct Bedrock ids stay distinct + assert bedrock_model_key("amazon.titan-text-001") == "titantext001" + assert bedrock_model_key("amazon.titan-text-002") == "titantext002" + assert bedrock_model_key("amazon.titan-text-001") != bedrock_model_key("amazon.titan-text-002") + # real dated/versioned ids are unaffected + assert bedrock_model_key("anthropic.claude-haiku-4-5-20251001-v1:0") == "claudehaiku45" + assert bedrock_model_key("eu.anthropic.claude-sonnet-4-6") == "claudesonnet46" + + +def test_unparseable_markup_keeps_the_cost_instead_of_zeroing_it() -> None: + """Defence in depth, and cross-port parity — `coerce_markup` means neither + branch is reachable through `emit()` (see the coerce test further down). + + The two bad inputs are not interchangeable: a bad COST leaves nothing to bill, + but a bad MARKUP only loses the multiplier, and returning 0 for it would + discard a good cost. JS already fell back to 1.0 here; Python returned "0", so + the ports would have billed differently had anything reached it. + """ + assert apply_markup("0.0042", "1.5") == "0.0063" + for bad in ("abc", "", "1,5", "None"): + assert apply_markup("0.0042", bad) == "0.0042", f"markup={bad!r} must not zero the cost" + # an unparseable COST is different: there is nothing to bill + assert apply_markup("abc", "1.5") == "0" + + def test_openrouter_miss_returns_none() -> None: table = parse_openrouter(_OPENROUTER_RAW) assert lookup_openrouter(table, "anthropic", "totally-made-up-model") is None @@ -447,6 +504,15 @@ def test_deoverlapped_token_total(usage: CanonicalUsage, expected: int, why: str assert deoverlapped_token_total(usage) == expected, why +@pytest.mark.parametrize("provider", ["openai", "workers-ai"]) +def test_openai_shaped_providers_treat_reasoning_as_a_subset(provider: str) -> None: + """workers-ai is reached ONLY through Cloudflare's OpenAI-compatible endpoint, so + reasoning is a subset of output there exactly as it is for real OpenAI. Omitting it + from _OUTPUT_INCLUDES_REASONING counted the subset twice — 1900 against 1100.""" + u = CanonicalUsage(input=100, output=1000, reasoning=800, model="m", provider=provider, api="chat") + assert deoverlapped_token_total(u) == 1100 + + def test_precomputed_unit_matches_the_split_path_basis() -> None: """The two cost branches must report the same quantity for one call — that was the actual complaint: `unit` on the single-event path used a different basis @@ -1346,6 +1412,38 @@ def test_warm_pricing_closes_the_cold_start_race() -> None: assert all(e["code"] == "llm_cost" for e in flat) # priced, not a token-event fallback +def test_bad_markup_is_coerced_to_one_reported_and_still_bills_the_cost() -> None: + """`markup` is customer input (`extra_lago={"markup": ...}`), so a comma decimal + like "1,5" genuinely arrives. `coerce_markup` is the guard that catches it; this + pins the end-to-end consequence, which no test covered: the cost is still billed + at 1.0 rather than zeroed, and the lost markup reaches on_error.""" + seen: list[tuple[Exception, str]] = [] + sdk, received = _price_sdk(_warm_provider(), on_error=lambda e, c: seen.append((e, c))) + try: + u = CanonicalUsage( + input=1000, output=500, model="claude-opus-4-8", provider="anthropic", api="native" + ) + # annotated `float | None`, but it arrives from untyped customer input + sdk.emit(u, markup="1,5") # type: ignore[arg-type] + assert sdk.flush(timeout=2.0) + finally: + sdk.shutdown(timeout=1.0) + + events = _by_token_type(received) + assert events, "a bad markup must not drop the cost events" + for token_type, ev in events.items(): + props = ev["properties"] + assert props["markup"] == "1", f"{token_type}: bad markup should coerce to 1.0" + assert props["value"] == props["base_cost"], ( + f"{token_type}: should bill the un-marked-up cost, not {props['value']!r}" + ) + assert Decimal(props["value"]) > 0, f"{token_type}: a bad markup must not zero the bill" + + contexts = [c for _, c in seen] + assert "pricing" in contexts, f"the invalid markup must reach on_error; got {contexts}" + assert any("markup" in str(e) and "1,5" in str(e) for e, _ in seen) + + def test_price_mode_emits_one_event_per_token_type() -> None: """A real per-field breakdown (OpenRouter has both input/output prices for this model) splits into one llm_cost event per token_type, so Lago's @@ -1702,3 +1800,22 @@ def test_default_mode_is_tokens_unchanged() -> None: sdk.shutdown(timeout=1.0) flat = [e for batch in received for e in batch] assert {e["code"] for e in flat} == {"llm_input_tokens", "llm_output_tokens"} + + +def test_workers_ai_compat_prefix_is_defined_exactly_once() -> None: + """Two unrelated layers must agree on this string: `adapters/openai_native` + decides the PROVIDER from it, `pricing` strips it before a catalog lookup. They + must never import each other, so it lives in `canonical`. A drift between two + copies is a silently unpriced call, not a crash — which is why this is asserted + rather than left to review.""" + src = pathlib.Path(__file__).resolve().parents[2] / "src" / "lago_agent_sdk" + definitions = [ + f"{path.relative_to(src)}:{i}" + for path in src.rglob("*.py") + for i, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1) + if re.match(r"\s*_?WORKERS_AI_COMPAT_PREFIX\s*=", line) + ] + assert definitions == ["canonical.py:19"] or len(definitions) == 1, ( + f"expected one definition, found {definitions}" + ) + assert WORKERS_AI_COMPAT_PREFIX == "workers-ai/" diff --git a/tests/unit/test_queue.py b/tests/unit/test_queue.py index b4805ee..0423f6b 100644 --- a/tests/unit/test_queue.py +++ b/tests/unit/test_queue.py @@ -2,6 +2,9 @@ from __future__ import annotations +import pathlib +import subprocess +import sys import threading import time @@ -178,6 +181,70 @@ def sender(batch): q.shutdown(timeout=2.0) +_REENTRANT_OVERFLOW_PROGRAM = """ +import sys, threading +sys.path.insert(0, {src!r}) +from lago_agent_sdk.queue import EventQueue + +calls = {{"n": 0}} +def on_error(exc, where): + calls["n"] += 1 + if calls["n"] > 200: # runaway guard so this exits rather than spinning + raise SystemExit(3) + q.push({{"diagnostic": True}}) # re-enters push() from inside the hook + +q = EventQueue(sender=lambda b: None, flush_interval=10.0, max_batch_size=1000, + max_buffer_size=1, on_error=on_error) +for i in range(3): + q.push({{"i": i}}) +q.shutdown(timeout=1.0) +print("OK", calls["n"]) +""" + + +def _run_reentrant_overflow(timeout: float = 20.0) -> subprocess.CompletedProcess: + """Run the re-entrant-overflow scenario in a SUBPROCESS. + + It has to be a subprocess: once the deadlock happens, the wedged producer holds + `_lock` forever, and `EventQueue.__init__`'s `atexit` shutdown then blocks on that + same lock at interpreter exit. The process is poisoned, so an in-process test + would hang the whole session instead of reporting a failure. Out-of-process, a + hang is just a timeout we can assert on. + """ + src = str(pathlib.Path(__file__).resolve().parents[2] / "src") + return subprocess.run( + [sys.executable, "-c", _REENTRANT_OVERFLOW_PROGRAM.format(src=src)], + capture_output=True, + text=True, + timeout=timeout, + ) + + +def test_overflow_report_does_not_deadlock_a_reentrant_callback(): + """The report must run with the lock RELEASED, and must not re-enter unboundedly. + + Two failure modes, one scenario. `_lock` is a plain Lock, not an RLock, so + reporting inside it meant a customer `on_error` that touched the SDK at all — + emitting a diagnostic, forcing a flush — blocked forever on a lock its own thread + already held; overflow happens under sustained load, which is exactly when such a + hook fires. Moving the report out of the lock then exposed the other half: the + buffer is full again by the time the hook runs, so a hook that pushes overflows + again and re-enters without bound. + """ + try: + proc = _run_reentrant_overflow() + except subprocess.TimeoutExpired: + raise AssertionError( + "re-entrant on_error during overflow hung the process — the report is holding the lock" + ) from None + assert proc.returncode == 0, f"exit {proc.returncode}: " + ( + "runaway re-entrant reporting" if proc.returncode == 3 else proc.stderr[-400:] + ) + assert proc.stdout.startswith("OK"), proc.stdout + reports = int(proc.stdout.split()[1]) + assert reports <= 10, f"expected a bounded number of overflow reports, got {reports}" + + def test_overflow_is_reported_through_on_error(): """An overflow drops BILLABLE events. It was logger.warning only, so a customer watching on_error never learned revenue had been lost; the JS port already @@ -206,6 +273,36 @@ 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]) +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. + + Treating them as transient re-prepended the identical batch at the head of the FIFO + and backed off to 60s forever, blocking everything behind it. Routing them to + `_send_individually` splits the batch and delivers what is deliverable — which is + what the isolation path was built for, and it was unreachable for exactly the batch + that most needed it. + """ + sent_individually: list = [] + + def sender(batch): + if len(batch) > 1: + raise LagoApiError(status, "batch too large / unacceptable as-is") + sent_individually.append(batch[0]["id"]) # each event succeeds alone + + q = EventQueue(sender=sender, flush_interval=0.05, max_batch_size=10, max_retry_seconds=0.5) + try: + for i in range(4): + q.push({"id": i}) + assert q.flush(timeout=3.0), "queue should drain, not head-of-line block" + assert sorted(sent_individually) == [0, 1, 2, 3], ( + f"every event should have been delivered individually, got {sent_individually}" + ) + assert q._backoff_seconds == 0.0, "splitting must not leave a stale backoff" + finally: + q.shutdown(timeout=1.0) + + @pytest.mark.parametrize("status", [429, 408]) def test_throttling_4xx_is_retried_not_dropped(status: int): """A rate-limited or timed-out batch must reach Lago eventually. Dropping @@ -363,3 +460,23 @@ def slow(batch): finally: blocking.set() q.shutdown(timeout=2.0) + + +def test_isolated_retries_keep_their_fifo_order() -> None: + """`_replay_failed` PREPENDS, so calling it once per event inside the isolation + loop reversed the survivors: a 413 batch of a,b,c,d,e whose b,c,d fail + transiently while isolated came back as d,c,b. FIFO is the queue's contract — + it is what makes oldest-dropped-first overflow and Lago's own event ordering + mean anything — so a recovery path must not silently invert it.""" + + def sender(batch): + if batch[0]["id"] in ("b", "c", "d"): + raise LagoApiError(503, "transient while isolated") + + q = EventQueue(sender=sender, flush_interval=60.0, max_batch_size=10, max_buffer_size=100) + try: + q._send_individually([{"id": i} for i in "abcde"], LagoApiError(413, "too large")) + with q._lock: + assert [e["id"] for e in q._buffer] == ["b", "c", "d"] + finally: + q.shutdown(timeout=1.0) diff --git a/tests/unit/test_sdk.py b/tests/unit/test_sdk.py index 65a8cdb..64711c1 100644 --- a/tests/unit/test_sdk.py +++ b/tests/unit/test_sdk.py @@ -2,6 +2,8 @@ from __future__ import annotations +import logging + import pytest from lago_agent_sdk import CanonicalUsage, LagoConfig, LagoSDK @@ -111,6 +113,19 @@ def test_explicit_api_url_still_wins_over_config(): sdk.shutdown(timeout=1.0) +@pytest.mark.parametrize("empty", ["", None]) +def test_empty_or_absent_api_url_keeps_the_production_default(empty): + """`api_url=os.environ.get("LAGO_API_URL", "")` with the var unset must NOT write + "". Downstream that is unrecoverable: requests raises MissingSchema, which is not a + LagoApiError, so the queue treats it as transient and retries at the 60s ceiling + forever — all billing stops with only a growing buffer as the symptom.""" + sdk = LagoSDK(api_key="k", api_url=empty) + try: + assert sdk.config.api_url == "https://api.getlago.com/api/v1" + finally: + sdk.shutdown(timeout=1.0) + + def test_default_api_url_is_still_production_when_nothing_is_passed(): """Changing the parameter default to None must not change this.""" sdk = LagoSDK(api_key="k") @@ -268,3 +283,73 @@ def test_caller_dimensions_win_on_a_collision_on_both_emitters(): # charged amount on a cost event, because `precise_total_amount_cents` is a # sibling of `properties`, not a member of it. assert cost[0]["precise_total_amount_cents"] == "1" + + +def test_negative_token_counts_are_reported_not_just_dropped(caplog) -> None: + """`CanonicalUsage` is exported and `emit()` takes one directly — the documented + way to backfill usage the SDK did not intercept — so a caller computing a delta + wrongly really can hand us a negative. Dropping it was correct (Lago would sum a + negative billable quantity) but it was the only drop path that never reached + on_error.""" + seen: list[tuple[Exception, str]] = [] + received: list = [] + cfg = LagoConfig(api_key="k", default_subscription_id="sub", on_error=lambda e, c: seen.append((e, c))) + sdk = LagoSDK(api_key="k", config=cfg) + sdk._queue._sender = lambda b: received.append(list(b)) # type: ignore[attr-defined] + try: + sdk.emit(CanonicalUsage(input=-100, output=50, model="m", provider="anthropic", api="native")) + assert sdk.flush(timeout=2.0) + finally: + sdk.shutdown(timeout=1.0) + + flat = [e for batch in received for e in batch] + values = [e["properties"]["value"] for e in flat] + assert all(not str(v).startswith("-") for v in values), f"a negative was billed: {values}" + assert "negative_tokens" in [c for _, c in seen], f"drop must reach on_error; got {seen}" + assert "input" in str(next(e for e, c in seen if c == "negative_tokens")) + + +def test_a_dropped_event_logs_exactly_once(caplog) -> None: + """`_report_error` invokes on_error AND logs. An extra logger.error alongside it + emitted the same drop twice under two levels, so a customer grepping logs counted + one lost call as two — while the JS port logged nothing at all for it.""" + cfg = LagoConfig(api_key="k") # no default subscription -> the drop path + sdk = LagoSDK(api_key="k", config=cfg) + try: + with caplog.at_level(logging.DEBUG, logger="lago_agent_sdk"): + sdk.emit(CanonicalUsage(input=10, output=5, model="m", provider="anthropic", api="native")) + drop_lines = [r for r in caplog.records if "subscription" in r.getMessage()] + assert len(drop_lines) == 1, f"expected one line per drop, got {[r.getMessage() for r in drop_lines]}" + finally: + sdk.shutdown(timeout=1.0) + + +def test_verify_ssl_false_survives_a_broken_urllib3(monkeypatch) -> None: + """Suppressing the InsecureRequestWarning is an optional convenience; it must + never be able to fail construction. This sits on an advertised path — `verify_ssl` + is a first-class constructor arg the docstring recommends for local dev — so an + ImportError/AttributeError here would take down `LagoSDK()` for exactly the setup + the flag exists to serve. The old code reached through `requests.packages`, a + legacy shim with no guarantee of existing.""" + import builtins + + real_import = builtins.__import__ + + def boom(name, *args, **kwargs): + if name == "urllib3": + raise ImportError("no urllib3 for you") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", boom) + # Also remove the legacy shim the old code reached through, so this test fails + # against that version instead of silently passing: `requests.packages` is a + # compatibility alias, not API, and nothing guarantees it is present. + import requests + + monkeypatch.delattr(requests, "packages", raising=False) + + sdk = LagoSDK(api_key="k", api_url="https://example.invalid/api/v1", verify_ssl=False) + try: + assert sdk.config.verify_ssl is False + finally: + sdk.shutdown(timeout=1.0)