Fix the queue, subscription and pricing paths flagged in review

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/<vendor>/<model>". 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"

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)

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:

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

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():