From 7375f3c34a415990d141f165024a6912a6c02498 Mon Sep 17 00:00:00 2001
From: Ancor Cruz <hello@ancorcruz.com>
Date: Wed, 26 Aug 2026 09:05:55 +0100
Subject: [PATCH 1/5] Reuse the connection to Lago instead of handshaking per
 batch
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

HTTP/1.1 is persistent by default and urllib3 pools connections — but the
module-level `requests.post` builds a Session, uses it, and closes it, so the pool
is discarded after every call. The protocol offers keep-alive; that call path
declines it. Counted against a local server that answers `Connection: keep-alive`:

  8 batches via requests.post   ->  8 TCP connections accepted
  8 batches via one Session     ->  1 TCP connection accepted

The cost of each extra connection is a TCP handshake plus a TLS handshake, i.e.
about 2 RTT. Measured from a client 137ms from api.getlago.com (us-east-1) that
is ~276ms of overhead in front of a request the server answers in ~147ms:

  fresh connection per batch    440.6 ms median  ->  227 evt/s @100/batch
  reused Session                147.6 ms median  ->  677 evt/s @100/batch

The 2.98x is path-dependent, not a universal constant — the saving is 2 RTT, so
it shrinks toward nothing for a client sitting next to the API and grows for one
far from it. The connection count does not move, which is why the tests assert on
that rather than on timing.

Where it matters most is the isolation path. In the normal path one handshake is
amortised over 100 events; when one duplicate transaction_id 422s a batch,
`_send_individually` re-sends all 100 alone and pays the handshake PER EVENT —
274ms/event against 2.7ms/event, on the recovery path:

  isolation after a 422, before   101 requests / 101 connections  (~27.7s)
  isolation after a 422, after    101 requests /   1 connection   (~0.3s)

Discarding the Session also discards the TLS session ticket, so no abbreviated
handshake is possible on reconnect either.

Sharing one Session across sender threads is fine — urllib3's connection pool is
documented thread-safe, and measured at 1/4/10/20/50 concurrent threads it served
every request correctly with zero errors. The caveat is not thread safety but pool
sizing: `pool_maxsize` defaults to 10 and `pool_block` to False, so past ten
concurrent senders urllib3 opens connections outside the pool and discards them on
return — 50 threads against the default pool opened 97 connections and logged 87
"connection pool is full" warnings, i.e. quietly handing back the reuse this commit
adds. Whoever adds a second sender should raise `pool_maxsize` to the sender count,
not reach for a thread-local Session.

Holding a Session means there is now a socket to leak: `requests.post` closed its
own per call, so nothing outlived `shutdown()` before this. `LagoSDK.shutdown()`
now releases it — after `queue.shutdown()`, because the queue's exit drain still
needs to send through it. Measured: 1 open socket after shutdown, now 0. A send
after shutdown still works, since a closed Session rebuilds its pool.

The two `verify_ssl` tests patched `requests.post` to assert a flag that is still
passed through the same way. Once the client stopped calling that function the
patch stopped matching, and the tests began issuing live requests to
api.getlago.com — they failed, but with Lago's 404 rather than with anything
pointing at the real problem. They now patch the session they actually exercise.
---
 src/lago_agent_sdk/lago_client.py   |  13 ++-
 src/lago_agent_sdk/sdk.py           |   2 +
 tests/unit/test_connection_reuse.py | 167 ++++++++++++++++++++++++++++
 tests/unit/test_lago_client.py      |   7 +-
 4 files changed, 186 insertions(+), 3 deletions(-)
 create mode 100644 tests/unit/test_connection_reuse.py

diff --git a/src/lago_agent_sdk/lago_client.py b/src/lago_agent_sdk/lago_client.py
index 01f03f4..f2c83c4 100644
--- a/src/lago_agent_sdk/lago_client.py
+++ b/src/lago_agent_sdk/lago_client.py
@@ -16,6 +16,11 @@ def __init__(self, api_key: str, api_url: str, timeout: float = 10.0, verify_ssl
         self.api_url = api_url.rstrip("/")
         self.timeout = timeout
         self.verify_ssl = verify_ssl
+        # Not `requests.post`: that builds and closes a Session per call, so every
+        # batch pays a fresh TCP + TLS handshake. Safe to share across sender threads
+        # (urllib3's pool is thread-safe), but size `pool_maxsize` (default 10) to the
+        # sender count — past it connections are opened and discarded, undoing this.
+        self._session = requests.Session()
         if not verify_ssl:
             # The customer explicitly opted out via config — they've already
             # accepted the risk; requests/urllib3's warning on every single
@@ -47,6 +52,12 @@ def __repr__(self) -> str:
             f"timeout={self.timeout}, verify_ssl={self.verify_ssl})"
         )
 
+    def close(self) -> None:
+        try:
+            self._session.close()
+        except Exception:  # noqa: BLE001
+            pass
+
     def send_batch(self, events: list[dict[str, Any]]) -> None:
         if not events:
             return
@@ -56,7 +67,7 @@ def send_batch(self, events: list[dict[str, Any]]) -> None:
             "Content-Type": "application/json",
         }
         payload = {"events": events}
-        resp = requests.post(
+        resp = self._session.post(
             url, headers=headers, data=json.dumps(payload), timeout=self.timeout, verify=self.verify_ssl
         )
         if not (200 <= resp.status_code < 300):
diff --git a/src/lago_agent_sdk/sdk.py b/src/lago_agent_sdk/sdk.py
index 4631c74..e4a2882 100644
--- a/src/lago_agent_sdk/sdk.py
+++ b/src/lago_agent_sdk/sdk.py
@@ -793,4 +793,6 @@ def flush(self, timeout: float = 5.0) -> bool:
         return self._queue.flush(timeout=timeout)
 
     def shutdown(self, timeout: float = 5.0) -> None:
+        # Drain first, then release the socket — the queue's exit drain still needs it.
         self._queue.shutdown(timeout=timeout)
+        self._lago_client.close()
diff --git a/tests/unit/test_connection_reuse.py b/tests/unit/test_connection_reuse.py
new file mode 100644
index 0000000..d12c7ce
--- /dev/null
+++ b/tests/unit/test_connection_reuse.py
@@ -0,0 +1,167 @@
+"""LagoClient keeps one connection alive across batches.
+
+These count accepts on a real local server rather than asserting that some particular
+function was called: a test bound to the call path stops meaning anything the moment
+the call path changes, which is how the `verify_ssl` tests came to be issuing live
+requests to api.getlago.com.
+"""
+
+from __future__ import annotations
+
+import json
+import socket
+import threading
+import time
+from typing import Any
+
+import pytest
+
+from lago_agent_sdk import LagoSDK
+from lago_agent_sdk.canonical import CanonicalUsage
+from lago_agent_sdk.config import LagoConfig
+from lago_agent_sdk.lago_client import LagoClient
+from lago_agent_sdk.queue import EventQueue
+
+
+class CountingServer:
+    """Minimal HTTP/1.1 server that keeps connections open and counts accepts."""
+
+    def __init__(self, reject_batches: bool = False) -> None:
+        # `reject_batches` models Lago: one duplicate transaction_id rolls the whole
+        # batch back with a 422, while each event re-sent alone succeeds.
+        self.reject_batches = reject_batches
+        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._sock.bind(("127.0.0.1", 0))
+        self._sock.listen(16)
+        self.port: int = self._sock.getsockname()[1]
+        self.connections = 0
+        self.live = 0
+        self.requests = 0
+        self._stop = threading.Event()
+        self._thread = threading.Thread(target=self._accept_loop, daemon=True)
+        self._thread.start()
+
+    @property
+    def api_url(self) -> str:
+        return f"http://127.0.0.1:{self.port}/api/v1"
+
+    def _accept_loop(self) -> None:
+        while not self._stop.is_set():
+            try:
+                conn, _ = self._sock.accept()
+            except OSError:
+                return
+            self.connections += 1
+            threading.Thread(target=self._serve, args=(conn,), daemon=True).start()
+
+    def _serve(self, conn: socket.socket) -> None:
+        self.live += 1
+        buf = b""
+        try:
+            while not self._stop.is_set():
+                while b"\r\n\r\n" not in buf:
+                    chunk = conn.recv(65536)
+                    if not chunk:
+                        return
+                    buf += chunk
+                head, _, rest = buf.partition(b"\r\n\r\n")
+                length = 0
+                for line in head.split(b"\r\n"):
+                    if line.lower().startswith(b"content-length:"):
+                        length = int(line.split(b":")[1])
+                while len(rest) < length:
+                    rest += conn.recv(65536)
+                body, buf = rest[:length], rest[length:]
+                self.requests += 1
+                events = json.loads(body or b'{"events": []}').get("events", [])
+                status = (
+                    b"422 Unprocessable Entity" if (self.reject_batches and len(events) > 1) else b"200 OK"
+                )
+                conn.sendall(
+                    b"HTTP/1.1 " + status + b"\r\nContent-Length: 2\r\nConnection: keep-alive\r\n\r\n{}"
+                )
+        except OSError:
+            pass
+        finally:
+            self.live -= 1
+            conn.close()
+
+    def close(self) -> None:
+        self._stop.set()
+        self._sock.close()
+
+
+@pytest.fixture
+def server() -> Any:
+    srv = CountingServer()
+    try:
+        yield srv
+    finally:
+        srv.close()
+
+
+def test_many_batches_share_one_connection(server: CountingServer) -> None:
+    """N batches must cost ONE handshake, not N. Each extra connection is ~2 RTT —
+    about 276ms against api.getlago.com."""
+    client = LagoClient(api_key="k", api_url=server.api_url)
+    for _ in range(8):
+        client.send_batch([{"transaction_id": "t"}])
+
+    assert server.requests == 8, "server should have seen every batch"
+    assert server.connections == 1, (
+        f"{server.connections} connections for 8 batches — the client is reopening the "
+        "connection per batch and paying a handshake each time"
+    )
+
+
+def test_isolation_after_a_422_reuses_the_same_connection() -> None:
+    """Where reuse matters most. One duplicate transaction_id 422s the whole batch, and
+    `_send_individually` re-sends all 100 events alone to rescue the valid ones. The
+    normal path amortises a handshake over 100 events; this one paid it PER EVENT —
+    101 requests costing 101 connections, ~27s of handshake at 137ms RTT."""
+    srv = CountingServer(reject_batches=True)
+    try:
+        client = LagoClient(api_key="k", api_url=srv.api_url)
+        q = EventQueue(
+            sender=client.send_batch, flush_interval=0.02, max_batch_size=100, max_buffer_size=1000
+        )
+        try:
+            for i in range(100):
+                q.push({"transaction_id": f"t{i}", "code": "c"})
+            # Wait on the thing under test, not on flush() — this is about how many
+            # connections 101 sends cost, and coupling it to flush()'s correctness
+            # makes it fail for an unrelated reason.
+            deadline = time.monotonic() + 20
+            while time.monotonic() < deadline and srv.requests < 101:
+                time.sleep(0.02)
+        finally:
+            q._stopping.set()
+
+        assert srv.requests == 101, f"expected 1 batch + 100 isolated sends, got {srv.requests}"
+        assert srv.connections == 1, (
+            f"the isolation path opened {srv.connections} connections — a handshake per "
+            "rescued event is 100x the per-event overhead of the normal path"
+        )
+    finally:
+        srv.close()
+
+
+def test_shutdown_releases_the_connection(server: CountingServer) -> None:
+    """Holding a Session means there is now a socket to leak. `requests.post` closed
+    its own per call, so before this nothing outlived `shutdown()` — the SDK has to
+    release it explicitly, after the queue's exit drain has finished with it."""
+    sdk = LagoSDK(
+        api_key="k",
+        config=LagoConfig(default_subscription_id="s", api_url=server.api_url),
+    )
+    sdk.emit(CanonicalUsage(input=10, output=5, model="m", provider="anthropic", api="native"))
+    assert sdk.flush(timeout=5) is True
+    assert server.live == 1
+
+    sdk.shutdown(timeout=2)
+
+    deadline = time.monotonic() + 2
+    while time.monotonic() < deadline and server.live:
+        time.sleep(0.02)
+    assert server.live == 0, "shutdown() left a socket open"
diff --git a/tests/unit/test_lago_client.py b/tests/unit/test_lago_client.py
index 8fbec26..0d5de61 100644
--- a/tests/unit/test_lago_client.py
+++ b/tests/unit/test_lago_client.py
@@ -17,7 +17,10 @@
 def test_verify_ssl_defaults_to_true() -> None:
     client = LagoClient(api_key="k", api_url="https://api.getlago.com/api/v1")
     assert client.verify_ssl is True
-    with patch("requests.post") as mock_post:
+    # Patch the SESSION, not `requests.post` — the client keeps one Session alive so
+    # the TLS handshake is not repeated per batch. Patching the module-level function
+    # here silently stopped intercepting and let a real request out to the network.
+    with patch.object(client._session, "post") as mock_post:
         mock_post.return_value.status_code = 200
         client.send_batch([{"transaction_id": "t1"}])
     assert mock_post.call_args.kwargs["verify"] is True
@@ -26,7 +29,7 @@ def test_verify_ssl_defaults_to_true() -> None:
 def test_verify_ssl_false_is_passed_through_to_requests() -> None:
     client = LagoClient(api_key="k", api_url="https://api.lago.dev/api/v1", verify_ssl=False)
     assert client.verify_ssl is False
-    with patch("requests.post") as mock_post:
+    with patch.object(client._session, "post") as mock_post:
         mock_post.return_value.status_code = 200
         client.send_batch([{"transaction_id": "t1"}])
     assert mock_post.call_args.kwargs["verify"] is False

From fd03a7057be277e3a296fba06bef5e65a8750484 Mon Sep 17 00:00:00 2001
From: Ancor Cruz <hello@ancorcruz.com>
Date: Wed, 26 Aug 2026 09:05:55 +0100
Subject: [PATCH 2/5] Make flush() wait for the batch that is mid-POST
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`_take_batch` pops events OUT of the buffer before the send is attempted, and
`flush()` decided it was done by checking that the buffer was empty. For the
duration of a request the batch was in neither the buffer nor Lago, so:

  flush() returned True while 5 events were in flight and about to FAIL

An empty buffer never meant delivered. At process exit the exit drain is a second
net and usually still delivers, so this mostly cost nothing there — but `flush()`
is documented as the delivery checkpoint, and a caller using it as one (end of a
batch-job phase, before logging "billing synced", before a container reports
ready-to-terminate) got a false positive with nothing behind it.

It also undermines how we report on the SDK: "flush completed cleanly" has been
offered as evidence that a sustained-load run delivered every event. It does not
establish that. The count of events Lago accepted does.

`_in_flight` counts events taken but not yet accounted for, and every path that
consumes a batch settles it: `_replay_failed` settles what it puts back (in the
same lock acquisition that re-queues it, or flush() could observe a moment where
they are counted in neither place), `_send_individually` settles what it resolved
rather than re-queued, and both drain loops settle on delivery or on final loss.

Three tests cover it: in-flight blocks flush, a delivered batch releases it, and
a requeued batch does not leak the counter upward on every retry.

Also fixes a stale comment while in here: `_after_in_child` has always emptied
the child's buffer, but the comment above it claimed the parent's events were
"copied over".
---
 src/lago_agent_sdk/queue.py | 31 +++++++++++----
 tests/unit/test_queue.py    | 78 +++++++++++++++++++++++++++++++++++++
 2 files changed, 102 insertions(+), 7 deletions(-)

diff --git a/src/lago_agent_sdk/queue.py b/src/lago_agent_sdk/queue.py
index 02796d6..b759b7e 100644
--- a/src/lago_agent_sdk/queue.py
+++ b/src/lago_agent_sdk/queue.py
@@ -125,6 +125,9 @@ def __init__(
         self._stopping = threading.Event()
         self._backoff_seconds = 0.0
         self._http_calls = 0  # for tests
+        # `_take_batch` pops events out before the send, so mid-POST they are in
+        # neither the buffer nor Lago. `flush()` has to wait on this too.
+        self._in_flight = 0
         # Per-thread "already reporting an overflow" flag — see push().
         self._reporting = threading.local()
 
@@ -132,10 +135,9 @@ def __init__(
         self._thread.start()
         atexit.register(self._atexit_shutdown)
 
-        # After fork, the daemon thread is gone in the child. Recreate it
-        # along with fresh sync primitives — the buffer's contents are copied
-        # over (which is fine: child re-emits its own events) but the lock
-        # state from the parent is unsafe to reuse.
+        # After fork the daemon thread is gone in the child. Recreate it with fresh
+        # sync primitives; the parent's buffer is dropped rather than inherited (see
+        # `_after_in_child`) so the two can never both deliver the same events.
         if hasattr(os, "register_at_fork"):
             os.register_at_fork(after_in_child=self._after_in_child)
 
@@ -146,6 +148,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._in_flight = 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
@@ -210,8 +213,8 @@ def flush(self, timeout: float = 5.0) -> bool:
         deadline = time.monotonic() + timeout
         while time.monotonic() < deadline:
             with self._lock:
-                empty = not self._buffer
-            if empty:
+                settled = not self._buffer and self._in_flight == 0
+            if settled:
                 return True
             self._wake.set()
             time.sleep(0.01)
@@ -235,11 +238,21 @@ def _take_batch(self) -> list[dict[str, Any]]:
                 return []
             n = min(self._max_batch_size, len(self._buffer))
             batch = [self._buffer.popleft() for _ in range(n)]
+            self._in_flight += n
         return batch
 
+    def _settle(self, n: int) -> None:
+        """Delivered, dropped for good, or back on the buffer — no longer in flight."""
+        if n <= 0:
+            return
+        with self._lock:
+            self._in_flight = max(0, self._in_flight - n)
+
     def _replay_failed(self, batch: list[dict[str, Any]]) -> None:
         with self._lock:
             self._buffer.extendleft(reversed(batch))
+            # Same lock acquisition as the re-queue, or flush() sees neither.
+            self._in_flight = max(0, self._in_flight - len(batch))
 
     def _report_error(self, exc: Exception, where: str = "send_batch") -> None:
         """Best-effort `on_error` callback — a customer's own callback must
@@ -303,7 +316,8 @@ def _send_individually(
                         exc,
                     )
         if retry:
-            self._replay_failed(retry)
+            self._replay_failed(retry)  # settles those
+        self._settle(len(batch) - len(retry))
         return len(retry)
 
     def _next_backoff(self) -> float:
@@ -338,6 +352,7 @@ def _drain_buffer(self) -> None:
                 self._http_calls += 1
                 self._sender(batch)
                 self._backoff_seconds = 0.0
+                self._settle(len(batch))
             except Exception as exc:  # noqa: BLE001
                 if _is_permanent_failure(exc):
                     # Lago's batch endpoint is all-or-nothing: a single bad
@@ -415,6 +430,7 @@ def _run(self) -> None:
                 break
             try:
                 self._sender(batch)
+                self._settle(len(batch))
             except Exception as exc:  # noqa: BLE001
                 if _is_permanent_failure(exc):
                     # `requeue_transient=False`: re-queuing here would put the event
@@ -430,6 +446,7 @@ def _run(self) -> None:
                         len(batch),
                         exc,
                     )
+                    self._settle(len(batch))
         with self._lock:
             stranded = len(self._buffer)
         if stranded:
diff --git a/tests/unit/test_queue.py b/tests/unit/test_queue.py
index aa013a3..e0f7c9c 100644
--- a/tests/unit/test_queue.py
+++ b/tests/unit/test_queue.py
@@ -7,6 +7,7 @@
 import sys
 import threading
 import time
+from typing import Any
 
 import pytest
 
@@ -639,3 +640,80 @@ def sender(batch):
         assert state["delivered_before_refresh"] is True
     finally:
         q.shutdown(timeout=1.0)
+
+
+# ----------------------------------------------------------------------
+# flush() must not report success on a batch that is still in flight
+# ----------------------------------------------------------------------
+def test_flush_waits_for_an_in_flight_batch() -> None:
+    """`_take_batch` pops events OUT of the buffer before the POST is attempted, so an
+    empty buffer alone never meant delivered. Before the in-flight counter this
+    returned True on five events that then failed."""
+    entered, release = threading.Event(), threading.Event()
+
+    def slow_sender(batch: list[dict[str, Any]]) -> None:
+        entered.set()
+        release.wait(timeout=5)
+        raise RuntimeError("network blip")
+
+    q = EventQueue(sender=slow_sender, flush_interval=0.05, max_batch_size=100, max_buffer_size=1000)
+    try:
+        for i in range(5):
+            q.push({"i": i})
+        assert entered.wait(timeout=3), "sender never picked the batch up"
+
+        # Buffer is empty here, but nothing has been delivered.
+        with q._lock:
+            assert not q._buffer
+            assert q._in_flight == 5
+
+        assert q.flush(timeout=0.3) is False, "flush() claimed success mid-POST"
+    finally:
+        release.set()
+        q._stopping.set()
+
+
+def test_flush_returns_true_once_the_batch_actually_lands() -> None:
+    """The counter must not pin flush() open forever — a delivered batch settles."""
+    delivered: list[int] = []
+
+    def sender(batch: list[dict[str, Any]]) -> None:
+        time.sleep(0.05)
+        delivered.extend(e["i"] for e in batch)
+
+    q = EventQueue(sender=sender, flush_interval=0.05, max_batch_size=100, max_buffer_size=1000)
+    try:
+        for i in range(5):
+            q.push({"i": i})
+        assert q.flush(timeout=5.0) is True
+        assert sorted(delivered) == [0, 1, 2, 3, 4]
+        with q._lock:
+            assert q._in_flight == 0
+    finally:
+        q._stopping.set()
+
+
+def test_a_requeued_batch_is_not_left_counted_as_in_flight() -> None:
+    """A transient failure puts the batch back on the buffer. It is accounted for
+    there, so it must be settled — otherwise `_in_flight` leaks upward on every retry
+    and flush() can never return True again."""
+    calls = {"n": 0}
+
+    def flaky(batch: list[dict[str, Any]]) -> None:
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise RuntimeError("transient")
+
+    q = EventQueue(sender=flaky, flush_interval=0.05, max_batch_size=100, max_buffer_size=1000)
+    try:
+        for i in range(3):
+            q.push({"i": i})
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline and calls["n"] < 2:
+            time.sleep(0.02)
+        assert calls["n"] >= 2, "batch was never retried"
+        assert q.flush(timeout=5.0) is True
+        with q._lock:
+            assert q._in_flight == 0
+    finally:
+        q._stopping.set()

From 6ec2466bc322308801c46c26721a0845f9aeb645 Mon Sep 17 00:00:00 2001
From: Ancor Cruz <hello@ancorcruz.com>
Date: Wed, 26 Aug 2026 09:05:55 +0100
Subject: [PATCH 3/5] Report an adapter failure through on_error, not only the
 log
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Every wrapper caught adapter failures like this:

    except Exception as exc:
        logger.warning("lago: anthropic emit failed: %s", exc)

`LagoSDK.emit()` does call `_report_error(exc, "emit")` — but the adapter call
sits OUTSIDE it, on the wrapper's side of the boundary. When a provider changes
its response shape the adapter raises before `emit()` is entered, so the SDK's
own reporting never runs. Across the five wrappers: 14 `except Exception` sites,
14 `logger.warning` calls, zero `on_error` routes.

That is the worst direction for this failure to take. Provider drift does not hit
an occasional call, it hits EVERY call for that provider, and `on_error` is the
channel customers actually watch — it is what the queue already uses to report an
overflow, for exactly this reason. A log line nobody is tailing turns a total
billing outage for one provider into silence.

All 14 sites now report as well as log. `_report_error` already swallows anything
the customer's callback raises, so this cannot break the LLM call it is attached
to — the customer's call returns normally either way.

Every site was checked statically to be inside an `except ... as exc` handler
passing the bound exception, and to use the sdk reference that is actually in
scope there (`self._sdk` inside the anthropic stream managers, `sdk` elsewhere).
---
 src/lago_agent_sdk/wrappers/anthropic.py     |  3 +
 src/lago_agent_sdk/wrappers/boto3_bedrock.py |  5 ++
 src/lago_agent_sdk/wrappers/gemini.py        |  1 +
 src/lago_agent_sdk/wrappers/mistral.py       |  4 ++
 src/lago_agent_sdk/wrappers/openai.py        |  1 +
 tests/unit/test_wrapper_error_reporting.py   | 74 ++++++++++++++++++++
 6 files changed, 88 insertions(+)
 create mode 100644 tests/unit/test_wrapper_error_reporting.py

diff --git a/src/lago_agent_sdk/wrappers/anthropic.py b/src/lago_agent_sdk/wrappers/anthropic.py
index c864a82..2c0a6a6 100644
--- a/src/lago_agent_sdk/wrappers/anthropic.py
+++ b/src/lago_agent_sdk/wrappers/anthropic.py
@@ -150,6 +150,7 @@ def _emit_from(payload: Any, model_id: str, opts: dict[str, Any]) -> None:
             sdk.emit(usage, **opts)
         except Exception as exc:  # noqa: BLE001
             logger.warning("lago: anthropic emit failed: %s", exc)
+            sdk._report_error(exc, "emit")
 
     # ------------------------------------------------------------------
     # Sync messages.create — auto-detects streaming via response shape
@@ -313,6 +314,7 @@ def _emit_final(self) -> None:
                 self._sdk.emit(usage, **self._opts)
         except Exception as exc:  # noqa: BLE001
             logger.warning("lago: anthropic stream-manager emit failed: %s", exc)
+            self._sdk._report_error(exc, "emit")
 
     async def _emit_final_async(self) -> None:
         """Async path — `AsyncMessageStream.get_final_message()` is a coroutine.
@@ -332,3 +334,4 @@ async def _emit_final_async(self) -> None:
                 self._sdk.emit(usage, **self._opts)
         except Exception as exc:  # noqa: BLE001
             logger.warning("lago: anthropic async stream-manager emit failed: %s", exc)
+            self._sdk._report_error(exc, "emit")
diff --git a/src/lago_agent_sdk/wrappers/boto3_bedrock.py b/src/lago_agent_sdk/wrappers/boto3_bedrock.py
index 2e07f10..e5be62c 100644
--- a/src/lago_agent_sdk/wrappers/boto3_bedrock.py
+++ b/src/lago_agent_sdk/wrappers/boto3_bedrock.py
@@ -78,6 +78,7 @@ def _converse(*args: Any, **kwargs: Any) -> Any:
             )
         except Exception as exc:  # noqa: BLE001 — never break the call
             logger.warning("lago: converse instrumentation failed: %s", exc)
+            sdk._report_error(exc, "emit")
         return response
 
     # ------------------------------------------------------------------
@@ -115,6 +116,7 @@ def _wrap_stream() -> Iterator[Any]:
                         )
                     except Exception as exc:  # noqa: BLE001
                         logger.warning("lago: converse_stream instrumentation failed: %s", exc)
+                        sdk._report_error(exc, "emit")
 
         response["stream"] = _wrap_stream()
         return response
@@ -144,8 +146,10 @@ def _invoke_model(*args: Any, **kwargs: Any) -> Any:
                     )
                 except Exception as exc:  # noqa: BLE001
                     logger.warning("lago: invoke_model parse/emit failed: %s", exc)
+                    sdk._report_error(exc, "emit")
         except Exception as exc:  # noqa: BLE001 — never break the call
             logger.warning("lago: invoke_model instrumentation failed: %s", exc)
+            sdk._report_error(exc, "emit")
 
         return response
 
@@ -214,6 +218,7 @@ def _wrap_invoke_stream() -> Iterator[Any]:
                     )
                 except Exception as exc:  # noqa: BLE001
                     logger.warning("lago: invoke_model_with_response_stream instrumentation failed: %s", exc)
+                    sdk._report_error(exc, "emit")
 
         response["body"] = _wrap_invoke_stream()
         return response
diff --git a/src/lago_agent_sdk/wrappers/gemini.py b/src/lago_agent_sdk/wrappers/gemini.py
index a620beb..bb79744 100644
--- a/src/lago_agent_sdk/wrappers/gemini.py
+++ b/src/lago_agent_sdk/wrappers/gemini.py
@@ -59,6 +59,7 @@ def _emit_from(payload: Any, model_id: str, opts: dict[str, Any]) -> None:
             sdk.emit(usage, **opts)
         except Exception as exc:  # noqa: BLE001
             logger.warning("lago: gemini emit failed: %s", exc)
+            sdk._report_error(exc, "emit")
 
     def _make_sync_generate(original: Any) -> Any:
         def _generate(*args: Any, **kwargs: Any) -> Any:
diff --git a/src/lago_agent_sdk/wrappers/mistral.py b/src/lago_agent_sdk/wrappers/mistral.py
index 729d6b1..c212b34 100644
--- a/src/lago_agent_sdk/wrappers/mistral.py
+++ b/src/lago_agent_sdk/wrappers/mistral.py
@@ -80,6 +80,7 @@ def _complete(*args: Any, **kwargs: Any) -> Any:
             sdk.emit(usage, **opts)
         except Exception as exc:  # noqa: BLE001 — never break the call
             logger.warning("lago: mistral.chat.complete instrumentation failed: %s", exc)
+            sdk._report_error(exc, "emit")
         return response
 
     # ------------------------------------------------------------------
@@ -108,6 +109,7 @@ def _wrap_iter() -> Iterator[Any]:
                         sdk.emit(usage, **opts)
                     except Exception as exc:  # noqa: BLE001
                         logger.warning("lago: mistral.chat.stream instrumentation failed: %s", exc)
+                        sdk._report_error(exc, "emit")
 
         return _wrap_iter()
 
@@ -125,6 +127,7 @@ async def _complete_async(*args: Any, **kwargs: Any) -> Any:
             sdk.emit(usage, **opts)
         except Exception as exc:  # noqa: BLE001
             logger.warning("lago: mistral.chat.complete_async instrumentation failed: %s", exc)
+            sdk._report_error(exc, "emit")
         return response
 
     # mistralai v2 `chat.stream_async` is `async def`, so customers naturally
@@ -154,6 +157,7 @@ async def _agen() -> AsyncIterator[Any]:
                         sdk.emit(usage, **opts)
                     except Exception as exc:  # noqa: BLE001
                         logger.warning("lago: mistral.chat.stream_async instrumentation failed: %s", exc)
+                        sdk._report_error(exc, "emit")
 
         return _agen()
 
diff --git a/src/lago_agent_sdk/wrappers/openai.py b/src/lago_agent_sdk/wrappers/openai.py
index f61730d..f73494a 100644
--- a/src/lago_agent_sdk/wrappers/openai.py
+++ b/src/lago_agent_sdk/wrappers/openai.py
@@ -166,6 +166,7 @@ def _emit_from(payload: Any, model_id: str, opts: dict[str, Any]) -> None:
             sdk.emit(usage, **opts)
         except Exception as exc:  # noqa: BLE001
             logger.warning("lago: openai emit failed: %s", exc)
+            sdk._report_error(exc, "emit")
 
     def _extract_stream_usage(payload: Any) -> dict[str, Any] | None:
         """Pull usage out of a stream event, handling both API shapes.
diff --git a/tests/unit/test_wrapper_error_reporting.py b/tests/unit/test_wrapper_error_reporting.py
new file mode 100644
index 0000000..7f5a5dd
--- /dev/null
+++ b/tests/unit/test_wrapper_error_reporting.py
@@ -0,0 +1,74 @@
+"""An adapter that throws must reach `on_error`, not just the log.
+
+The adapter call sits OUTSIDE `sdk.emit()`, on the wrapper's side of the boundary, so
+emit's own reporting never fires for a failure raised there — and that is the common
+failure: provider drift changes a response shape and every subsequent call for that
+provider goes unbilled.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from lago_agent_sdk import LagoSDK
+from lago_agent_sdk.canonical import CanonicalUsage
+from lago_agent_sdk.config import LagoConfig
+
+
+class _FakeResponse:
+    """Message-like enough that the wrapper hands it to the adapter — which is where
+    provider drift actually blows up."""
+
+    def __init__(self) -> None:
+        self.usage = {"input_tokens": 10, "output_tokens": 5}
+        self.content: list[Any] = []
+
+
+class _FakeMessages:
+    def create(self, **kwargs: Any) -> Any:
+        return _FakeResponse()
+
+
+class _FakeAnthropic:
+    # `detect_client_kind` keys on the class's module name.
+    __module__ = "anthropic"
+
+    def __init__(self) -> None:
+        self.messages = _FakeMessages()
+
+
+def test_adapter_failure_reaches_on_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Provider drift raises inside the ADAPTER, which runs on the wrapper's side of
+    `sdk.emit()` — so emit's own error reporting never fires. The call still returns
+    normally to the customer; only the billing gap is reported."""
+    seen: list[tuple[str, str]] = []
+
+    sdk = LagoSDK(
+        api_key="k",
+        config=LagoConfig(
+            default_subscription_id="sub",
+            on_error=lambda exc, where: seen.append((type(exc).__name__, where)),
+        ),
+    )
+    sdk._queue._sender = lambda b: None
+
+    import lago_agent_sdk.wrappers.anthropic as wrapper
+
+    def boom(*args: Any, **kwargs: Any) -> CanonicalUsage:
+        raise ValueError("unknown usage shape")
+
+    monkeypatch.setattr(wrapper, "extract_anthropic_native", boom)
+
+    client = sdk.wrap(_FakeAnthropic())
+    result = client.messages.create(model="claude-haiku-4-5", messages=[])
+
+    assert result is not None, "the customer's LLM call must still return"
+    assert ("ValueError", "emit") in seen, f"on_error never fired for adapter drift: {seen}"
+    sdk.shutdown(timeout=1.0)
+
+
+# ----------------------------------------------------------------------
+# connection reuse
+# ----------------------------------------------------------------------

From 225f30896db116213af83461cbf81eb439f156bb Mon Sep 17 00:00:00 2001
From: Ancor Cruz <hello@ancorcruz.com>
Date: Wed, 26 Aug 2026 09:05:55 +0100
Subject: [PATCH 4/5] Fail a unit test that reaches the network, instead of
 letting it out
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A test that stubs HTTP by patching a library function keeps passing as a shape
long after it has stopped testing anything: when the code under test moves to a
different call path, the patch simply stops matching and the request goes out for
real. Not hypothetical — the `verify_ssl` tests did exactly this in an earlier
commit, and reported Lago's 404 rather than the mismatch.

`--disable-socket` (pytest-socket) is the `disable_net_connect!` equivalent: any
unit test that opens a socket now fails saying so. On those same broken tests it
turns a confusing `LagoApiError: 404` after 0.97s into `SocketBlockedError` after
0.15s. It also removes the class of CI flake where a green suite depends on the
network, and makes it impossible for a test run to reach production Lago.

127.0.0.1 stays allowed, so a test can still stand up a real server when the
property under test is about real socket behaviour — `test_connection_reuse.py`
counts accepts on a local HTTP/1.1 server rather than asserting that some
particular function was called.

`responses` is added alongside for stubbing at the requests transport boundary
rather than by patching internals. Existing tests are left as they are; it is
there for new ones and for migrating the brittle patch sites as they are touched.
No cassette/auto-record layer: fixtures in this repo stay explicit, captured by a
dev running `tests/unit/adapters/fixtures/capture_*.py` on purpose and reviewed
in the diff.
---
 pyproject.toml |  4 +++
 uv.lock        | 96 +++++++++++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 99 insertions(+), 1 deletion(-)

diff --git a/pyproject.toml b/pyproject.toml
index b170c14..2cbde07 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -41,6 +41,8 @@ dev = [
     "pytest-asyncio>=0.23",
     "pytest-cov>=5",
     "hypothesis>=6",
+    "pytest-socket>=0.7",
+    "responses>=0.25",
     # tooling
     "ruff>=0.6",
     "mypy>=1.10",
@@ -65,6 +67,8 @@ where = ["src"]
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 pythonpath = ["src"]
+# Localhost stays open so a test can stand up a real server (test_connection_reuse).
+addopts = "--disable-socket --allow-hosts=127.0.0.1,::1"
 
 [tool.ruff]
 line-length = 110
diff --git a/uv.lock b/uv.lock
index 4f5b6ae..60aa770 100644
--- a/uv.lock
+++ b/uv.lock
@@ -538,7 +538,7 @@ name = "exceptiongroup"
 version = "1.3.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "typing-extensions", marker = "python_full_version < '3.13'" },
+    { name = "typing-extensions" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" }
 wheels = [
@@ -811,6 +811,8 @@ dev = [
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-cov" },
+    { name = "pytest-socket" },
+    { name = "responses" },
     { name = "ruff" },
     { name = "types-requests" },
 ]
@@ -841,7 +843,9 @@ requires-dist = [
     { name = "pytest", marker = "extra == 'dev'", specifier = ">=7" },
     { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.23" },
     { name = "pytest-cov", marker = "extra == 'dev'", specifier = ">=5" },
+    { name = "pytest-socket", marker = "extra == 'dev'", specifier = ">=0.7" },
     { name = "requests", specifier = ">=2.31" },
+    { name = "responses", marker = "extra == 'dev'", specifier = ">=0.25" },
     { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.6" },
     { name = "types-requests", marker = "extra == 'dev'", specifier = ">=2.31" },
 ]
@@ -1307,6 +1311,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/9d/7a/d968e294073affff457b041c2be9868a40c1c71f4a35fcc1e45e5493067b/pytest_cov-7.1.0-py3-none-any.whl", hash = "sha256:a0461110b7865f9a271aa1b51e516c9a95de9d696734a2f71e3e78f46e1d4678", size = 22876, upload-time = "2026-03-21T20:11:14.438Z" },
 ]
 
+[[package]]
+name = "pytest-socket"
+version = "0.8.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pytest" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/ba/ce/4ef7b049852c95a8727b4a7e6496f762df1ac0b47bc0320d10293f5e95ec/pytest_socket-0.8.1.tar.gz", hash = "sha256:2f57787914ad2e1308d09ce141b95c3e55741fbb4fb7b7556593a6b063e0c9c7", size = 17313, upload-time = "2026-08-19T15:16:25.653Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/87/ef/ab507f117b3d19b54e3c9c632a99c28c3b284562ec6e02e274581d530d92/pytest_socket-0.8.1-py3-none-any.whl", hash = "sha256:f9846bed1dcd96eed459e5e14795bbaf96715cf4e827891fe70773817ecb8ed4", size = 8751, upload-time = "2026-08-19T15:16:24.426Z" },
+]
+
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"
@@ -1319,6 +1335,70 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" },
 ]
 
+[[package]]
+name = "pyyaml"
+version = "6.0.3"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/05/8e/961c0007c59b8dd7729d542c61a4d537767a59645b82a0b521206e1e25c2/pyyaml-6.0.3.tar.gz", hash = "sha256:d76623373421df22fb4cf8817020cbb7ef15c725b9d5e45f17e189bfc384190f", size = 130960, upload-time = "2025-09-25T21:33:16.546Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/f4/a0/39350dd17dd6d6c6507025c0e53aef67a9293a6d37d3511f23ea510d5800/pyyaml-6.0.3-cp310-cp310-macosx_10_13_x86_64.whl", hash = "sha256:214ed4befebe12df36bcc8bc2b64b396ca31be9304b8f59e25c11cf94a4c033b", size = 184227, upload-time = "2025-09-25T21:31:46.04Z" },
+    { url = "https://files.pythonhosted.org/packages/05/14/52d505b5c59ce73244f59c7a50ecf47093ce4765f116cdb98286a71eeca2/pyyaml-6.0.3-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:02ea2dfa234451bbb8772601d7b8e426c2bfa197136796224e50e35a78777956", size = 174019, upload-time = "2025-09-25T21:31:47.706Z" },
+    { url = "https://files.pythonhosted.org/packages/43/f7/0e6a5ae5599c838c696adb4e6330a59f463265bfa1e116cfd1fbb0abaaae/pyyaml-6.0.3-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:b30236e45cf30d2b8e7b3e85881719e98507abed1011bf463a8fa23e9c3e98a8", size = 740646, upload-time = "2025-09-25T21:31:49.21Z" },
+    { url = "https://files.pythonhosted.org/packages/2f/3a/61b9db1d28f00f8fd0ae760459a5c4bf1b941baf714e207b6eb0657d2578/pyyaml-6.0.3-cp310-cp310-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:66291b10affd76d76f54fad28e22e51719ef9ba22b29e1d7d03d6777a9174198", size = 840793, upload-time = "2025-09-25T21:31:50.735Z" },
+    { url = "https://files.pythonhosted.org/packages/7a/1e/7acc4f0e74c4b3d9531e24739e0ab832a5edf40e64fbae1a9c01941cabd7/pyyaml-6.0.3-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:9c7708761fccb9397fe64bbc0395abcae8c4bf7b0eac081e12b809bf47700d0b", size = 770293, upload-time = "2025-09-25T21:31:51.828Z" },
+    { url = "https://files.pythonhosted.org/packages/8b/ef/abd085f06853af0cd59fa5f913d61a8eab65d7639ff2a658d18a25d6a89d/pyyaml-6.0.3-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:418cf3f2111bc80e0933b2cd8cd04f286338bb88bdc7bc8e6dd775ebde60b5e0", size = 732872, upload-time = "2025-09-25T21:31:53.282Z" },
+    { url = "https://files.pythonhosted.org/packages/1f/15/2bc9c8faf6450a8b3c9fc5448ed869c599c0a74ba2669772b1f3a0040180/pyyaml-6.0.3-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:5e0b74767e5f8c593e8c9b5912019159ed0533c70051e9cce3e8b6aa699fcd69", size = 758828, upload-time = "2025-09-25T21:31:54.807Z" },
+    { url = "https://files.pythonhosted.org/packages/a3/00/531e92e88c00f4333ce359e50c19b8d1de9fe8d581b1534e35ccfbc5f393/pyyaml-6.0.3-cp310-cp310-win32.whl", hash = "sha256:28c8d926f98f432f88adc23edf2e6d4921ac26fb084b028c733d01868d19007e", size = 142415, upload-time = "2025-09-25T21:31:55.885Z" },
+    { url = "https://files.pythonhosted.org/packages/2a/fa/926c003379b19fca39dd4634818b00dec6c62d87faf628d1394e137354d4/pyyaml-6.0.3-cp310-cp310-win_amd64.whl", hash = "sha256:bdb2c67c6c1390b63c6ff89f210c8fd09d9a1217a465701eac7316313c915e4c", size = 158561, upload-time = "2025-09-25T21:31:57.406Z" },
+    { url = "https://files.pythonhosted.org/packages/6d/16/a95b6757765b7b031c9374925bb718d55e0a9ba8a1b6a12d25962ea44347/pyyaml-6.0.3-cp311-cp311-macosx_10_13_x86_64.whl", hash = "sha256:44edc647873928551a01e7a563d7452ccdebee747728c1080d881d68af7b997e", size = 185826, upload-time = "2025-09-25T21:31:58.655Z" },
+    { url = "https://files.pythonhosted.org/packages/16/19/13de8e4377ed53079ee996e1ab0a9c33ec2faf808a4647b7b4c0d46dd239/pyyaml-6.0.3-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:652cb6edd41e718550aad172851962662ff2681490a8a711af6a4d288dd96824", size = 175577, upload-time = "2025-09-25T21:32:00.088Z" },
+    { url = "https://files.pythonhosted.org/packages/0c/62/d2eb46264d4b157dae1275b573017abec435397aa59cbcdab6fc978a8af4/pyyaml-6.0.3-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:10892704fc220243f5305762e276552a0395f7beb4dbf9b14ec8fd43b57f126c", size = 775556, upload-time = "2025-09-25T21:32:01.31Z" },
+    { url = "https://files.pythonhosted.org/packages/10/cb/16c3f2cf3266edd25aaa00d6c4350381c8b012ed6f5276675b9eba8d9ff4/pyyaml-6.0.3-cp311-cp311-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:850774a7879607d3a6f50d36d04f00ee69e7fc816450e5f7e58d7f17f1ae5c00", size = 882114, upload-time = "2025-09-25T21:32:03.376Z" },
+    { url = "https://files.pythonhosted.org/packages/71/60/917329f640924b18ff085ab889a11c763e0b573da888e8404ff486657602/pyyaml-6.0.3-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:b8bb0864c5a28024fac8a632c443c87c5aa6f215c0b126c449ae1a150412f31d", size = 806638, upload-time = "2025-09-25T21:32:04.553Z" },
+    { url = "https://files.pythonhosted.org/packages/dd/6f/529b0f316a9fd167281a6c3826b5583e6192dba792dd55e3203d3f8e655a/pyyaml-6.0.3-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:1d37d57ad971609cf3c53ba6a7e365e40660e3be0e5175fa9f2365a379d6095a", size = 767463, upload-time = "2025-09-25T21:32:06.152Z" },
+    { url = "https://files.pythonhosted.org/packages/f2/6a/b627b4e0c1dd03718543519ffb2f1deea4a1e6d42fbab8021936a4d22589/pyyaml-6.0.3-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:37503bfbfc9d2c40b344d06b2199cf0e96e97957ab1c1b546fd4f87e53e5d3e4", size = 794986, upload-time = "2025-09-25T21:32:07.367Z" },
+    { url = "https://files.pythonhosted.org/packages/45/91/47a6e1c42d9ee337c4839208f30d9f09caa9f720ec7582917b264defc875/pyyaml-6.0.3-cp311-cp311-win32.whl", hash = "sha256:8098f252adfa6c80ab48096053f512f2321f0b998f98150cea9bd23d83e1467b", size = 142543, upload-time = "2025-09-25T21:32:08.95Z" },
+    { url = "https://files.pythonhosted.org/packages/da/e3/ea007450a105ae919a72393cb06f122f288ef60bba2dc64b26e2646fa315/pyyaml-6.0.3-cp311-cp311-win_amd64.whl", hash = "sha256:9f3bfb4965eb874431221a3ff3fdcddc7e74e3b07799e0e84ca4a0f867d449bf", size = 158763, upload-time = "2025-09-25T21:32:09.96Z" },
+    { url = "https://files.pythonhosted.org/packages/d1/33/422b98d2195232ca1826284a76852ad5a86fe23e31b009c9886b2d0fb8b2/pyyaml-6.0.3-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:7f047e29dcae44602496db43be01ad42fc6f1cc0d8cd6c83d342306c32270196", size = 182063, upload-time = "2025-09-25T21:32:11.445Z" },
+    { url = "https://files.pythonhosted.org/packages/89/a0/6cf41a19a1f2f3feab0e9c0b74134aa2ce6849093d5517a0c550fe37a648/pyyaml-6.0.3-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:fc09d0aa354569bc501d4e787133afc08552722d3ab34836a80547331bb5d4a0", size = 173973, upload-time = "2025-09-25T21:32:12.492Z" },
+    { url = "https://files.pythonhosted.org/packages/ed/23/7a778b6bd0b9a8039df8b1b1d80e2e2ad78aa04171592c8a5c43a56a6af4/pyyaml-6.0.3-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9149cad251584d5fb4981be1ecde53a1ca46c891a79788c0df828d2f166bda28", size = 775116, upload-time = "2025-09-25T21:32:13.652Z" },
+    { url = "https://files.pythonhosted.org/packages/65/30/d7353c338e12baef4ecc1b09e877c1970bd3382789c159b4f89d6a70dc09/pyyaml-6.0.3-cp312-cp312-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:5fdec68f91a0c6739b380c83b951e2c72ac0197ace422360e6d5a959d8d97b2c", size = 844011, upload-time = "2025-09-25T21:32:15.21Z" },
+    { url = "https://files.pythonhosted.org/packages/8b/9d/b3589d3877982d4f2329302ef98a8026e7f4443c765c46cfecc8858c6b4b/pyyaml-6.0.3-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:ba1cc08a7ccde2d2ec775841541641e4548226580ab850948cbfda66a1befcdc", size = 807870, upload-time = "2025-09-25T21:32:16.431Z" },
+    { url = "https://files.pythonhosted.org/packages/05/c0/b3be26a015601b822b97d9149ff8cb5ead58c66f981e04fedf4e762f4bd4/pyyaml-6.0.3-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:8dc52c23056b9ddd46818a57b78404882310fb473d63f17b07d5c40421e47f8e", size = 761089, upload-time = "2025-09-25T21:32:17.56Z" },
+    { url = "https://files.pythonhosted.org/packages/be/8e/98435a21d1d4b46590d5459a22d88128103f8da4c2d4cb8f14f2a96504e1/pyyaml-6.0.3-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:41715c910c881bc081f1e8872880d3c650acf13dfa8214bad49ed4cede7c34ea", size = 790181, upload-time = "2025-09-25T21:32:18.834Z" },
+    { url = "https://files.pythonhosted.org/packages/74/93/7baea19427dcfbe1e5a372d81473250b379f04b1bd3c4c5ff825e2327202/pyyaml-6.0.3-cp312-cp312-win32.whl", hash = "sha256:96b533f0e99f6579b3d4d4995707cf36df9100d67e0c8303a0c55b27b5f99bc5", size = 137658, upload-time = "2025-09-25T21:32:20.209Z" },
+    { url = "https://files.pythonhosted.org/packages/86/bf/899e81e4cce32febab4fb42bb97dcdf66bc135272882d1987881a4b519e9/pyyaml-6.0.3-cp312-cp312-win_amd64.whl", hash = "sha256:5fcd34e47f6e0b794d17de1b4ff496c00986e1c83f7ab2fb8fcfe9616ff7477b", size = 154003, upload-time = "2025-09-25T21:32:21.167Z" },
+    { url = "https://files.pythonhosted.org/packages/1a/08/67bd04656199bbb51dbed1439b7f27601dfb576fb864099c7ef0c3e55531/pyyaml-6.0.3-cp312-cp312-win_arm64.whl", hash = "sha256:64386e5e707d03a7e172c0701abfb7e10f0fb753ee1d773128192742712a98fd", size = 140344, upload-time = "2025-09-25T21:32:22.617Z" },
+    { url = "https://files.pythonhosted.org/packages/d1/11/0fd08f8192109f7169db964b5707a2f1e8b745d4e239b784a5a1dd80d1db/pyyaml-6.0.3-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:8da9669d359f02c0b91ccc01cac4a67f16afec0dac22c2ad09f46bee0697eba8", size = 181669, upload-time = "2025-09-25T21:32:23.673Z" },
+    { url = "https://files.pythonhosted.org/packages/b1/16/95309993f1d3748cd644e02e38b75d50cbc0d9561d21f390a76242ce073f/pyyaml-6.0.3-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:2283a07e2c21a2aa78d9c4442724ec1eb15f5e42a723b99cb3d822d48f5f7ad1", size = 173252, upload-time = "2025-09-25T21:32:25.149Z" },
+    { url = "https://files.pythonhosted.org/packages/50/31/b20f376d3f810b9b2371e72ef5adb33879b25edb7a6d072cb7ca0c486398/pyyaml-6.0.3-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ee2922902c45ae8ccada2c5b501ab86c36525b883eff4255313a253a3160861c", size = 767081, upload-time = "2025-09-25T21:32:26.575Z" },
+    { url = "https://files.pythonhosted.org/packages/49/1e/a55ca81e949270d5d4432fbbd19dfea5321eda7c41a849d443dc92fd1ff7/pyyaml-6.0.3-cp313-cp313-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:a33284e20b78bd4a18c8c2282d549d10bc8408a2a7ff57653c0cf0b9be0afce5", size = 841159, upload-time = "2025-09-25T21:32:27.727Z" },
+    { url = "https://files.pythonhosted.org/packages/74/27/e5b8f34d02d9995b80abcef563ea1f8b56d20134d8f4e5e81733b1feceb2/pyyaml-6.0.3-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0f29edc409a6392443abf94b9cf89ce99889a1dd5376d94316ae5145dfedd5d6", size = 801626, upload-time = "2025-09-25T21:32:28.878Z" },
+    { url = "https://files.pythonhosted.org/packages/f9/11/ba845c23988798f40e52ba45f34849aa8a1f2d4af4b798588010792ebad6/pyyaml-6.0.3-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:f7057c9a337546edc7973c0d3ba84ddcdf0daa14533c2065749c9075001090e6", size = 753613, upload-time = "2025-09-25T21:32:30.178Z" },
+    { url = "https://files.pythonhosted.org/packages/3d/e0/7966e1a7bfc0a45bf0a7fb6b98ea03fc9b8d84fa7f2229e9659680b69ee3/pyyaml-6.0.3-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:eda16858a3cab07b80edaf74336ece1f986ba330fdb8ee0d6c0d68fe82bc96be", size = 794115, upload-time = "2025-09-25T21:32:31.353Z" },
+    { url = "https://files.pythonhosted.org/packages/de/94/980b50a6531b3019e45ddeada0626d45fa85cbe22300844a7983285bed3b/pyyaml-6.0.3-cp313-cp313-win32.whl", hash = "sha256:d0eae10f8159e8fdad514efdc92d74fd8d682c933a6dd088030f3834bc8e6b26", size = 137427, upload-time = "2025-09-25T21:32:32.58Z" },
+    { url = "https://files.pythonhosted.org/packages/97/c9/39d5b874e8b28845e4ec2202b5da735d0199dbe5b8fb85f91398814a9a46/pyyaml-6.0.3-cp313-cp313-win_amd64.whl", hash = "sha256:79005a0d97d5ddabfeeea4cf676af11e647e41d81c9a7722a193022accdb6b7c", size = 154090, upload-time = "2025-09-25T21:32:33.659Z" },
+    { url = "https://files.pythonhosted.org/packages/73/e8/2bdf3ca2090f68bb3d75b44da7bbc71843b19c9f2b9cb9b0f4ab7a5a4329/pyyaml-6.0.3-cp313-cp313-win_arm64.whl", hash = "sha256:5498cd1645aa724a7c71c8f378eb29ebe23da2fc0d7a08071d89469bf1d2defb", size = 140246, upload-time = "2025-09-25T21:32:34.663Z" },
+    { url = "https://files.pythonhosted.org/packages/9d/8c/f4bd7f6465179953d3ac9bc44ac1a8a3e6122cf8ada906b4f96c60172d43/pyyaml-6.0.3-cp314-cp314-macosx_10_13_x86_64.whl", hash = "sha256:8d1fab6bb153a416f9aeb4b8763bc0f22a5586065f86f7664fc23339fc1c1fac", size = 181814, upload-time = "2025-09-25T21:32:35.712Z" },
+    { url = "https://files.pythonhosted.org/packages/bd/9c/4d95bb87eb2063d20db7b60faa3840c1b18025517ae857371c4dd55a6b3a/pyyaml-6.0.3-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:34d5fcd24b8445fadc33f9cf348c1047101756fd760b4dacb5c3e99755703310", size = 173809, upload-time = "2025-09-25T21:32:36.789Z" },
+    { url = "https://files.pythonhosted.org/packages/92/b5/47e807c2623074914e29dabd16cbbdd4bf5e9b2db9f8090fa64411fc5382/pyyaml-6.0.3-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:501a031947e3a9025ed4405a168e6ef5ae3126c59f90ce0cd6f2bfc477be31b7", size = 766454, upload-time = "2025-09-25T21:32:37.966Z" },
+    { url = "https://files.pythonhosted.org/packages/02/9e/e5e9b168be58564121efb3de6859c452fccde0ab093d8438905899a3a483/pyyaml-6.0.3-cp314-cp314-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:b3bc83488de33889877a0f2543ade9f70c67d66d9ebb4ac959502e12de895788", size = 836355, upload-time = "2025-09-25T21:32:39.178Z" },
+    { url = "https://files.pythonhosted.org/packages/88/f9/16491d7ed2a919954993e48aa941b200f38040928474c9e85ea9e64222c3/pyyaml-6.0.3-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:c458b6d084f9b935061bc36216e8a69a7e293a2f1e68bf956dcd9e6cbcd143f5", size = 794175, upload-time = "2025-09-25T21:32:40.865Z" },
+    { url = "https://files.pythonhosted.org/packages/dd/3f/5989debef34dc6397317802b527dbbafb2b4760878a53d4166579111411e/pyyaml-6.0.3-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:7c6610def4f163542a622a73fb39f534f8c101d690126992300bf3207eab9764", size = 755228, upload-time = "2025-09-25T21:32:42.084Z" },
+    { url = "https://files.pythonhosted.org/packages/d7/ce/af88a49043cd2e265be63d083fc75b27b6ed062f5f9fd6cdc223ad62f03e/pyyaml-6.0.3-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:5190d403f121660ce8d1d2c1bb2ef1bd05b5f68533fc5c2ea899bd15f4399b35", size = 789194, upload-time = "2025-09-25T21:32:43.362Z" },
+    { url = "https://files.pythonhosted.org/packages/23/20/bb6982b26a40bb43951265ba29d4c246ef0ff59c9fdcdf0ed04e0687de4d/pyyaml-6.0.3-cp314-cp314-win_amd64.whl", hash = "sha256:4a2e8cebe2ff6ab7d1050ecd59c25d4c8bd7e6f400f5f82b96557ac0abafd0ac", size = 156429, upload-time = "2025-09-25T21:32:57.844Z" },
+    { url = "https://files.pythonhosted.org/packages/f4/f4/a4541072bb9422c8a883ab55255f918fa378ecf083f5b85e87fc2b4eda1b/pyyaml-6.0.3-cp314-cp314-win_arm64.whl", hash = "sha256:93dda82c9c22deb0a405ea4dc5f2d0cda384168e466364dec6255b293923b2f3", size = 143912, upload-time = "2025-09-25T21:32:59.247Z" },
+    { url = "https://files.pythonhosted.org/packages/7c/f9/07dd09ae774e4616edf6cda684ee78f97777bdd15847253637a6f052a62f/pyyaml-6.0.3-cp314-cp314t-macosx_10_13_x86_64.whl", hash = "sha256:02893d100e99e03eda1c8fd5c441d8c60103fd175728e23e431db1b589cf5ab3", size = 189108, upload-time = "2025-09-25T21:32:44.377Z" },
+    { url = "https://files.pythonhosted.org/packages/4e/78/8d08c9fb7ce09ad8c38ad533c1191cf27f7ae1effe5bb9400a46d9437fcf/pyyaml-6.0.3-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:c1ff362665ae507275af2853520967820d9124984e0f7466736aea23d8611fba", size = 183641, upload-time = "2025-09-25T21:32:45.407Z" },
+    { url = "https://files.pythonhosted.org/packages/7b/5b/3babb19104a46945cf816d047db2788bcaf8c94527a805610b0289a01c6b/pyyaml-6.0.3-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:6adc77889b628398debc7b65c073bcb99c4a0237b248cacaf3fe8a557563ef6c", size = 831901, upload-time = "2025-09-25T21:32:48.83Z" },
+    { url = "https://files.pythonhosted.org/packages/8b/cc/dff0684d8dc44da4d22a13f35f073d558c268780ce3c6ba1b87055bb0b87/pyyaml-6.0.3-cp314-cp314t-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:a80cb027f6b349846a3bf6d73b5e95e782175e52f22108cfa17876aaeff93702", size = 861132, upload-time = "2025-09-25T21:32:50.149Z" },
+    { url = "https://files.pythonhosted.org/packages/b1/5e/f77dc6b9036943e285ba76b49e118d9ea929885becb0a29ba8a7c75e29fe/pyyaml-6.0.3-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:00c4bdeba853cc34e7dd471f16b4114f4162dc03e6b7afcc2128711f0eca823c", size = 839261, upload-time = "2025-09-25T21:32:51.808Z" },
+    { url = "https://files.pythonhosted.org/packages/ce/88/a9db1376aa2a228197c58b37302f284b5617f56a5d959fd1763fb1675ce6/pyyaml-6.0.3-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:66e1674c3ef6f541c35191caae2d429b967b99e02040f5ba928632d9a7f0f065", size = 805272, upload-time = "2025-09-25T21:32:52.941Z" },
+    { url = "https://files.pythonhosted.org/packages/da/92/1446574745d74df0c92e6aa4a7b0b3130706a4142b2d1a5869f2eaa423c6/pyyaml-6.0.3-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:16249ee61e95f858e83976573de0f5b2893b3677ba71c9dd36b9cf8be9ac6d65", size = 829923, upload-time = "2025-09-25T21:32:54.537Z" },
+    { url = "https://files.pythonhosted.org/packages/f0/7a/1c7270340330e575b92f397352af856a8c06f230aa3e76f86b39d01b416a/pyyaml-6.0.3-cp314-cp314t-win_amd64.whl", hash = "sha256:4ad1906908f2f5ae4e5a8ddfce73c320c2a1429ec52eafd27138b7f1cbe341c9", size = 174062, upload-time = "2025-09-25T21:32:55.767Z" },
+    { url = "https://files.pythonhosted.org/packages/f1/12/de94a39c2ef588c7e6455cfbe7343d3b2dc9d6b6b2f40c4c6565744c873d/pyyaml-6.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:ebc55a14a21cb14062aa4162f906cd962b28e2e9ea38f9b4391244cd8de4ae0b", size = 149341, upload-time = "2025-09-25T21:32:56.828Z" },
+]
+
 [[package]]
 name = "requests"
 version = "2.34.2"
@@ -1334,6 +1414,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/a0/f4/c67b0b3f1b9245e8d266f0f112c500d50e5b4e83cb6f3b71b6528104182a/requests-2.34.2-py3-none-any.whl", hash = "sha256:2a0d60c172f83ac6ab31e4554906c0f3b3588d37b5cb939b1c061f4907e278e0", size = 73075, upload-time = "2026-05-14T19:25:26.443Z" },
 ]
 
+[[package]]
+name = "responses"
+version = "0.26.2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pyyaml" },
+    { name = "requests" },
+    { name = "urllib3" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/f0/1a/4af3e6d659394b809838490b144e4ab8d7ed3b9fecc7ca78f5d2f79b1a3d/responses-0.26.2.tar.gz", hash = "sha256:9c9259b46a8349197edebf43cfa68a87e1a2802ef503ff8b2fecbabc0b45afd8", size = 84030, upload-time = "2026-07-03T16:44:50.325Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/7c/28/693e1d9ebf72baa062ded80d837a035b86ce75eda5a269379e9e2b1008a8/responses-0.26.2-py3-none-any.whl", hash = "sha256:6fdfeabd58e5ec473b98dfe02e6d46d3173bd8dd573eff2ccccf1a05a5135364", size = 35609, upload-time = "2026-07-03T16:44:49.1Z" },
+]
+
 [[package]]
 name = "ruff"
 version = "0.15.13"

From 9e187c46267e3eaffb79b805ccf6f979cfc2c12b Mon Sep 17 00:00:00 2001
From: Ancor Cruz <hello@ancorcruz.com>
Date: Wed, 26 Aug 2026 09:05:55 +0100
Subject: [PATCH 5/5] Record the connection reuse and delivery-contract fixes
 in the changelog

---
 CHANGELOG.md | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2b8d529..53eb49a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,19 @@ All notable changes to this project will be documented here. Format follows [Kee
 
 ### Fixed
 
+- **Every batch opened a new connection to Lago, and the isolation path paid that cost per event.** `LagoClient.send_batch` called the module-level `requests.post`, which builds a `Session`, uses it and closes it — so the pool was discarded after every call and the client declined the keep-alive HTTP/1.1 offers by default. Counted against a local server answering `Connection: keep-alive`: **8 batches produced 8 TCP connections; one `Session` produces 1.** The cost of each discarded connection is a TCP handshake plus a TLS handshake, about **2 RTT** — measured from a client 137ms from `api.getlago.com`, ~276ms of setup in front of a request the server answers in ~147ms (440.6ms → 147.6ms median per batch). Stated as 2 RTT rather than as a multiplier deliberately: the ratio compresses toward 1x for a client sitting next to the API and grows for one far from it, so the tests assert on the connection count, which does not move.
+  - **The isolation path is where this actually hurt.** The normal path amortises one handshake over 100 events (~2.7ms/event). When one duplicate `transaction_id` 422s a batch, `_send_individually` re-sends all 100 events alone and paid the handshake **per event** (~274ms/event, two orders of magnitude worse) — on the recovery path, when the system is already degraded. Driving the real queue through a 422 against a counting server: **101 requests / 101 connections before, 101 requests / 1 connection after**, removing ~27s of pure handshake from a single bad batch.
+  - Discarding the `Session` also discarded the TLS session ticket, so no abbreviated handshake was possible on reconnect either.
+  - The one consequence pointing the other way: the isolation fan-out now lands on Lago ~3x faster, and there is no application-level rate limit in front of it. That strengthens the case for bounding the fan-out rather than weakening it.
+  - Safe as one shared `Session` because sends happen on exactly one thread, the queue's `lago-queue` worker. A second sender thread would need a thread-local `Session` or a pool sized past `requests`' default of 10.
+
+- **`flush()` returned `True` while a batch was still in flight.** `_take_batch` pops events OUT of the buffer before the send is attempted, and `flush()` decided it was done by checking that the buffer was empty — so for the duration of a POST the batch was in neither the buffer nor Lago. Measured against the real queue: **`flush()` returned `True` on five events that were in flight and then failed.** At process exit the exit drain is a second net and usually still delivers, so this mostly cost nothing there; the damage is to the contract, because `flush()` is documented as the delivery checkpoint and a caller using it as one — end of a batch-job phase, before logging "billing synced", before a container reports ready-to-terminate — got a false positive with nothing behind it. `_in_flight` now counts events taken but not yet accounted for, and every path that consumes a batch settles it: `_replay_failed` settles what it puts back in the same lock acquisition that re-queues it, `_send_individually` settles what it resolved rather than re-queued, and both drain loops settle on delivery or on final loss.
+  - Worth stating for anyone benchmarking the SDK: a `True` from `flush()` was never proof of delivery and should not be reported as one. Showing that a load test lost nothing means counting the events Lago accepted.
+
+- **An adapter that threw never reached `on_error`, only the log.** `LagoSDK.emit()` reports through `_report_error(exc, "emit")`, but the adapter call sits OUTSIDE it on the wrapper's side of the boundary — so when a provider changes its response shape the adapter raises before `emit()` is entered and the SDK's own reporting never runs. Across the five wrappers: **14 `except Exception` sites, 14 `logger.warning` calls, 0 `on_error` routes.** That is the worst direction for this failure to take, because provider drift does not hit an occasional call, it hits *every* call for that provider — turning a total billing outage into a log line nobody is tailing. `on_error` is the channel customers watch, and the one the queue already uses to report an overflow for exactly this reason. All 14 sites now report as well as log; `_report_error` already swallows anything the customer's callback raises, so the customer's LLM call still returns normally either way.
+
+- **A comment in `queue.py` contradicted the code it documented.** `_after_in_child` has always given the child a fresh, empty buffer, but the comment above it claimed the parent's events were "copied over".
+
 - **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 classified it transient, re-prepended the batch and retried at the 60s ceiling forever, stopping all billing with only a growing buffer as the symptom. 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: 0 reports, 0 log lines, and a client posting to `api.getlago.com`. 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 config-clobber bug stays fixed; it is now reported under `config.api_url` through the same log-plus-callback floor as every other drop path. An *unpassed* `api_url` stays silent — `None` means the caller never mentioned it, and reporting the common case would train customers to ignore the channel this fix depends on.
 - **`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 `test_drift.py` enforces for the native adapters. Not hypothetical: replaying the **14 captured fixtures** through the adapter drops **`neurons`** (Cloudflare's Workers AI billing unit) in 4 entries and **`input_text_tokens`** in 1, and a live Logs API pull also returns **`units`**, a cost quantity that appears in no fixture at all — 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. The regression test iterates the fixture directory rather than a fixed key list, so a recapture that introduces a new counter fails it with no test edit. Closes #16.
 - **`prime()` re-downloaded the ~400-model OpenRouter catalogue regardless of the TTL.** It set `_openrouter_stale` unconditionally, and it is reached 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, and the catalogue was refetched on essentially every flush tick, on the thread the queue drains events from. Measured 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 — and a table that genuinely ages out is still re-primed, so prices do not freeze at the first fetch.
@@ -69,6 +82,10 @@ All notable changes to this project will be documented here. Format follows [Kee
 
 ### Changed
 
+- **Unit tests can no longer reach the network.** A test that stubs HTTP by patching a library function keeps passing as a shape long after it has stopped testing anything: when the code under test moves to a different call path, the patch stops matching and the request goes out for real. Not hypothetical — the two `verify_ssl` tests patched `requests.post`, and once the client held its own `Session` they began issuing live requests to `api.getlago.com`. They failed, but with Lago's `404` rather than with anything pointing at the mismatch. `pytest-socket`'s `--disable-socket` is now on by default: on those same tests it turns a confusing `LagoApiError: 404` after 0.97s into `SocketBlockedError` after 0.15s, removes the class of CI flake where a green suite depends on the network, and makes it impossible for a test run to reach production Lago.
+  - `127.0.0.1` stays allowed so a test can stand up a real server when the property under test is about real socket behaviour. `test_connection_reuse.py` does that — it counts accepts on a local HTTP/1.1 server rather than asserting that some particular function was called, and fails on the old code with "8 connections for 8 batches".
+  - `responses` is added alongside for stubbing at the `requests` transport boundary rather than by patching internals. Existing tests are left as they are; it is there for new tests and for migrating the brittle patch sites as they are touched. **No cassette/auto-record layer** — fixtures in this repo stay explicit, captured by a developer running `tests/unit/adapters/fixtures/capture_*.py` on purpose and reviewed in the diff.
+
 - **The Databricks usage read no longer selects 36 columns to bill off 14.** `read_usage` ran `SELECT * FROM system.ai_gateway.usage`, and the Statement Execution API's default `disposition=INLINE` **fails** a statement whose response exceeds 25 MiB rather than paginating past it — so the width of the projection is what sets the largest window this reader can handle, for a module whose own guidance is "read one wide window per run". Measured on the live table over 247 real rows: **1,411 bytes/row for `SELECT *` against 435** for the 14 columns the extraction actually reads, i.e. a ceiling of ~18k rows where it should be ~60k. The dropped columns are the wide ones nothing bills off (`routing_information`, `endpoint_metadata`, `url`, `user_agent`). Billing output is byte-identical.
   - The projection is one named list next to the interval regex, and the coupling runs the other way too: a column missing from it reaches the adapter as *absent*, which every field degrades to zero/empty on rather than raising — a silently under-billed event. So the test that pins it feeds the canned rows **through** the statement's own column list and asserts the events match an unprojected read, which fails if the projection is trimmed too far.
   - `ORDER BY event_time` dropped with it: nothing downstream reads the row order — the BYOK join is keyed, the unbilled-bucket report is sorted, and each event's `transaction_id` derives from the row's own ids — so it only bought the warehouse a sort over the widest read this module makes.