From 89840b0e0fb0319c358981e92eac23a29df939f2 Mon Sep 17 00:00:00 2001 From: Andres Lopez <190146319+AndresL230@users.noreply.github.com> Date: Wed, 22 Apr 2026 00:37:33 -0400 Subject: [PATCH 1/7] Add bucket overflow protection (Python mirror of Node 0.1.2) - Aggregator: introduce MAX_BUCKETS (2000) constant + max_buckets config field + would_overflow() method + max_buckets getter. Prevents silent data loss when a window crosses the ingest API's 422 threshold. Direct port of the Node SDK's overflow protection. - Public surface: re-export MAX_BUCKETS and FlushStatus from the package __init__. - Tests: cover the new aggregator behavior, the lastFlushStatus semantics on RecostHandle, and the transport-side chunking + rejection signaling paths introduced alongside this feature. --- recost/__init__.py | 5 +- recost/_aggregator.py | 34 ++++- tests/test_aggregator.py | 45 ++++++- tests/test_init.py | 14 ++ tests/test_transport.py | 274 ++++++++++++++++++++++++++++++++++++++- 5 files changed, 368 insertions(+), 4 deletions(-) diff --git a/recost/__init__.py b/recost/__init__.py index 7e818e2..6351c6e 100644 --- a/recost/__init__.py +++ b/recost/__init__.py @@ -6,6 +6,7 @@ """ from ._types import ( + FlushStatus, RecostConfig, MetricEntry, ProviderDef, @@ -16,7 +17,7 @@ from ._init import RecostHandle, init from ._provider_registry import BUILTIN_PROVIDERS, MatchResult, ProviderRegistry from ._interceptor import install, uninstall, is_installed -from ._aggregator import Aggregator +from ._aggregator import Aggregator, MAX_BUCKETS __all__ = [ "init", @@ -27,6 +28,7 @@ "ProviderDef", "RecostConfig", "TransportMode", + "FlushStatus", "ProviderRegistry", "BUILTIN_PROVIDERS", "MatchResult", @@ -34,4 +36,5 @@ "uninstall", "is_installed", "Aggregator", + "MAX_BUCKETS", ] diff --git a/recost/_aggregator.py b/recost/_aggregator.py index e9c2b14..151f2ac 100644 --- a/recost/_aggregator.py +++ b/recost/_aggregator.py @@ -16,6 +16,18 @@ from ._types import MetricEntry, RawEvent, WindowSummary +# --------------------------------------------------------------------------- +# Bucket cap — matches the ingest API's 422 threshold +# --------------------------------------------------------------------------- + +MAX_BUCKETS = 2000 +"""Maximum unique (provider, endpoint, method) triplets per window. + +Crossing this mid-window triggers an early flush so the current window is +preserved instead of silently dropped when the API returns 422. +""" + + # --------------------------------------------------------------------------- # Internal bucket structure # --------------------------------------------------------------------------- @@ -60,10 +72,12 @@ def __init__( project_id: str = "", environment: str = "development", sdk_version: str = "0.0.0", + max_buckets: int = MAX_BUCKETS, ) -> None: self._project_id = project_id self._environment = environment self._sdk_version = sdk_version + self._max_buckets = max_buckets self._buckets: Dict[str, _Bucket] = {} self._window_start: Optional[str] = None self._size = 0 @@ -72,6 +86,19 @@ def __init__( # Public API # --------------------------------------------------------------------------- + @staticmethod + def _key_for(event: RawEvent) -> str: + provider = event.provider if event.provider is not None else "unknown" + endpoint = event.endpoint_category if event.endpoint_category is not None else event.path + return f"{provider}::{endpoint}::{event.method}" + + def would_overflow(self, event: RawEvent) -> bool: + """True if ingesting ``event`` would allocate a new bucket while the + window is already at capacity. Callers should flush before ingesting.""" + if len(self._buckets) < self._max_buckets: + return False + return self._key_for(event) not in self._buckets + def ingest(self, event: RawEvent, cost_cents: float = 0.0) -> None: """Add one RawEvent to the current window.""" if self._window_start is None: @@ -79,7 +106,7 @@ def ingest(self, event: RawEvent, cost_cents: float = 0.0) -> None: provider = event.provider if event.provider is not None else "unknown" endpoint = event.endpoint_category if event.endpoint_category is not None else event.path - key = f"{provider}::{endpoint}::{event.method}" + key = self._key_for(event) bucket = self._buckets.get(key) if bucket is None: @@ -148,3 +175,8 @@ def size(self) -> int: def bucket_count(self) -> int: """Number of unique provider + endpoint + method groups.""" return len(self._buckets) + + @property + def max_buckets(self) -> int: + """Configured maximum buckets per window.""" + return self._max_buckets diff --git a/tests/test_aggregator.py b/tests/test_aggregator.py index e682dee..c83d984 100644 --- a/tests/test_aggregator.py +++ b/tests/test_aggregator.py @@ -6,7 +6,7 @@ from datetime import datetime, timezone -from recost._aggregator import Aggregator +from recost._aggregator import Aggregator, MAX_BUCKETS from recost._types import RawEvent @@ -337,3 +337,46 @@ def test_large_batch(self): assert len(summary.metrics) == 10 for entry in summary.metrics: assert entry.request_count == 100 + + +# --------------------------------------------------------------------------- +# Bucket overflow protection +# --------------------------------------------------------------------------- + +class TestBucketOverflow: + def test_max_buckets_constant_is_2000(self): + assert MAX_BUCKETS == 2000 + + def test_would_overflow_false_below_cap(self): + agg = Aggregator(max_buckets=10) + for i in range(5): + agg.ingest(make_event(provider=f"p{i}", endpoint_category=f"ep{i}")) + assert not agg.would_overflow(make_event(provider="new", endpoint_category="new")) + + def test_would_overflow_false_for_existing_key_at_cap(self): + agg = Aggregator(max_buckets=3) + agg.ingest(make_event(provider="a", endpoint_category="a")) + agg.ingest(make_event(provider="b", endpoint_category="b")) + agg.ingest(make_event(provider="c", endpoint_category="c")) + assert agg.bucket_count == 3 + # Same triplet — no new bucket needed + assert not agg.would_overflow(make_event(provider="a", endpoint_category="a")) + + def test_would_overflow_true_for_new_key_at_cap(self): + agg = Aggregator(max_buckets=3) + agg.ingest(make_event(provider="a", endpoint_category="a")) + agg.ingest(make_event(provider="b", endpoint_category="b")) + agg.ingest(make_event(provider="c", endpoint_category="c")) + assert agg.would_overflow(make_event(provider="d", endpoint_category="d")) + + def test_max_buckets_property(self): + assert Aggregator(max_buckets=500).max_buckets == 500 + assert Aggregator().max_buckets == MAX_BUCKETS + + def test_default_cap_fires_at_2001st_triplet(self): + agg = Aggregator() + for i in range(2000): + agg.ingest(make_event(provider=f"p{i}", endpoint_category=f"ep{i}")) + assert agg.bucket_count == 2000 + overflow = make_event(provider="p2000", endpoint_category="ep2000") + assert agg.would_overflow(overflow) diff --git a/tests/test_init.py b/tests/test_init.py index cec3945..1c58dc7 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -39,6 +39,20 @@ def test_dispose_is_idempotent(self): assert not is_installed() +class TestLastFlushStatus: + def test_none_before_any_flush(self): + handle = init(RecostConfig()) + try: + assert handle.last_flush_status is None + finally: + handle.dispose() + + def test_none_when_disabled(self): + handle = init(RecostConfig(enabled=False)) + assert handle.last_flush_status is None + handle.dispose() + + class TestExcludePatterns: def test_cloud_mode_excludes_base_url(self): # We can't easily test the filtering without making real requests, diff --git a/tests/test_transport.py b/tests/test_transport.py index 7d98154..e2e3c1d 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -2,13 +2,17 @@ Tests for recost/_transport.py """ +import asyncio import json +import logging +import socket import threading +import time from http.server import BaseHTTPRequestHandler, HTTPServer import pytest -from recost._transport import Transport, _post_cloud +from recost._transport import Transport, _LocalTransport, _post_cloud from recost._types import RecostConfig, MetricEntry, WindowSummary @@ -128,3 +132,271 @@ def test_no_retry_on_4xx(self, cloud_server): transport.dispose() # Should only have 1 request (no retries) assert len(_CloudHandler.received) == 1 + + +# --------------------------------------------------------------------------- +# Rejection signalling (422, on_error, warnings, last_flush_status) +# --------------------------------------------------------------------------- + + +def _make_metric(**overrides) -> MetricEntry: + defaults = dict( + provider="openai", + endpoint="chat_completions", + method="POST", + request_count=1, + error_count=0, + total_latency_ms=100, + p50_latency_ms=100, + p95_latency_ms=100, + total_request_bytes=10, + total_response_bytes=20, + estimated_cost_cents=1.0, + ) + defaults.update(overrides) + return MetricEntry(**defaults) + + +def _make_summary_with_metrics(metrics) -> WindowSummary: + return WindowSummary( + project_id="p", + environment="test", + sdk_language="python", + sdk_version="0.1.0", + window_start="2026-01-01T00:00:00Z", + window_end="2026-01-01T00:00:30Z", + metrics=metrics, + ) + + +class TestRejectionSignalling: + def test_422_fires_on_error_with_descriptive_message(self, cloud_server): + base_url, _ = cloud_server + _CloudHandler.response_code = 422 + + errors: list[Exception] = [] + config = RecostConfig( + api_key="k", + project_id="p", + base_url=base_url, + max_retries=0, + on_error=lambda e: errors.append(e), + debug=False, + ) + transport = Transport(config) + transport.send(_make_summary_with_metrics([_make_metric(), _make_metric(endpoint="b")])) + status = transport.last_flush_status + transport.dispose() + + assert len(errors) == 1 + assert "422" in str(errors[0]) + assert "windowSize=2" in str(errors[0]) + assert status is not None + assert status.status == "error" + assert status.window_size == 2 + + def test_422_logs_warning_when_debug_false(self, cloud_server, caplog): + base_url, _ = cloud_server + _CloudHandler.response_code = 422 + + config = RecostConfig( + api_key="k", + project_id="p", + base_url=base_url, + max_retries=0, + debug=False, + ) + transport = Transport(config) + with caplog.at_level(logging.WARNING, logger="recost"): + transport.send(_make_summary_with_metrics([_make_metric()])) + transport.dispose() + + rejection_records = [r for r in caplog.records if "HTTP 422" in r.getMessage()] + assert len(rejection_records) >= 1 + assert "windowSize=1" in rejection_records[0].getMessage() + + def test_last_flush_status_ok_on_success(self, cloud_server): + base_url, _ = cloud_server + _CloudHandler.response_code = 202 + config = RecostConfig(api_key="k", project_id="p", base_url=base_url, max_retries=0) + transport = Transport(config) + transport.send(_make_summary_with_metrics([_make_metric()])) + status = transport.last_flush_status + transport.dispose() + + assert status is not None + assert status.status == "ok" + assert status.window_size == 1 + assert isinstance(status.timestamp, int) + + def test_summaries_larger_than_max_buckets_are_chunked(self, cloud_server): + base_url, _ = cloud_server + _CloudHandler.response_code = 202 + config = RecostConfig( + api_key="k", + project_id="p", + base_url=base_url, + max_retries=0, + max_buckets=3, + ) + transport = Transport(config) + metrics = [_make_metric(endpoint=f"ep{i}") for i in range(7)] + transport.send(_make_summary_with_metrics(metrics)) + status = transport.last_flush_status + transport.dispose() + + assert len(_CloudHandler.received) == 3 # ceil(7/3) + sizes = [len(r["body"]["metrics"]) for r in _CloudHandler.received] + assert sizes == [3, 3, 1] + assert status is not None + assert status.status == "ok" + assert status.window_size == 1 # final chunk + + +# --------------------------------------------------------------------------- +# Local WebSocket transport — concurrency safety +# --------------------------------------------------------------------------- + + +def _find_free_port() -> int: + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + return port + + +class TestLocalTransportSync: + """Sync-path invariants: send() must never block, dispose() must always + unblock the loop thread, even when no WebSocket server is listening.""" + + def test_send_does_not_block_caller_without_server(self): + pytest.importorskip("websockets") + port = _find_free_port() + t = _LocalTransport(port=port) + try: + start = time.monotonic() + for _ in range(200): + t.send("payload") + elapsed = time.monotonic() - start + assert elapsed < 0.5, f"send() blocked for {elapsed:.2f}s" + finally: + t.dispose() + + def test_dispose_joins_cleanly_without_server(self): + pytest.importorskip("websockets") + port = _find_free_port() + t = _LocalTransport(port=port) + # Give the loop a moment to start and attempt its first connect + time.sleep(0.2) + start = time.monotonic() + t.dispose() + elapsed = time.monotonic() - start + assert t._thread is None + # Loop yields every 500ms, so dispose should return well under 2s + assert elapsed < 2.5, f"dispose() took {elapsed:.2f}s — loop was wedged" + + def test_send_after_dispose_is_noop(self): + """Calling send() post-dispose must not raise (loop may be closed).""" + pytest.importorskip("websockets") + port = _find_free_port() + t = _LocalTransport(port=port) + t.dispose() + t.send("late-payload") # must not raise + + def test_transport_local_mode_dispose_is_fast(self): + """Regression: the full Transport wrapper in local mode disposes + within the loop-yield window — no blocking queue.get hanging.""" + pytest.importorskip("websockets") + port = _find_free_port() + transport = Transport(RecostConfig(local_port=port)) + assert transport.mode == "local" + time.sleep(0.2) + start = time.monotonic() + transport.dispose() + assert time.monotonic() - start < 2.5 + + +class TestLocalTransportAsync: + """End-to-end: real WS server, exercises the reconnect path that was + prone to deadlocking when the drain loop used a blocking queue.get().""" + + async def test_reconnect_after_ws_drop_does_not_deadlock(self): + websockets = pytest.importorskip("websockets") + port = _find_free_port() + + received: list[str] = [] + connect_count = [0] + + async def handler(ws): + connect_count[0] += 1 + my_conn = connect_count[0] + try: + async for msg in ws: + received.append(msg) + # Drop the first connection after its first message to + # force the transport's reconnect path. + if my_conn == 1: + await ws.close() + return + except Exception: + pass + + server = await websockets.serve(handler, "127.0.0.1", port) + t = _LocalTransport(port=port) + try: + # Wait for initial connection + for _ in range(50): + if connect_count[0] >= 1: + break + await asyncio.sleep(0.05) + assert connect_count[0] >= 1, "transport never connected" + + t.send("msg-1") + for _ in range(100): + if len(received) >= 1: + break + await asyncio.sleep(0.05) + assert len(received) == 1 + + # Give the server's close() a beat to propagate to the client + await asyncio.sleep(0.1) + + # This send happens after the drop. The transport should re-queue + # it, reconnect, and deliver it on the new connection. + t.send("msg-2") + for _ in range(200): + if len(received) >= 2: + break + await asyncio.sleep(0.05) + + assert len(received) >= 2, f"reconnect failed, received={received}" + assert connect_count[0] >= 2, "transport did not reconnect" + + # Dispose must still exit promptly even after a reconnect cycle + start = time.monotonic() + t.dispose() + assert time.monotonic() - start < 2.5 + finally: + if t._thread is not None: + t.dispose() + server.close() + await server.wait_closed() + + async def test_send_from_running_event_loop_does_not_block(self): + """When the caller is already inside an asyncio loop (e.g. an async + web framework), send() must not block that loop. The transport's + own loop lives on a separate thread — this verifies the bridge.""" + pytest.importorskip("websockets") + port = _find_free_port() + t = _LocalTransport(port=port) + try: + start = time.monotonic() + for _ in range(100): + t.send("x") + # Yield so any accidental blocking would be visible + await asyncio.sleep(0) + elapsed = time.monotonic() - start + assert elapsed < 0.5 + finally: + t.dispose() From d1c7d6c4c97267341fc5031c502bd6dfc6c7f452 Mon Sep 17 00:00:00 2001 From: Andres Lopez <190146319+AndresL230@users.noreply.github.com> Date: Wed, 22 Apr 2026 00:37:42 -0400 Subject: [PATCH 2/7] Bump version to 0.1.2 --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 07ad8f9..6f47a91 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "recost" -version = "0.1.1" +version = "0.1.2" description = "Recost middleware for Python — API cost intelligence" readme = "README.md" license = { file = "LICENSE" } From fc326c70feaa1964c538c190c95b584153a1abd0 Mon Sep 17 00:00:00 2001 From: Andres Lopez <190146319+AndresL230@users.noreply.github.com> Date: Wed, 13 May 2026 02:48:32 -0400 Subject: [PATCH 3/7] Add AUDIT.md and ROADMAP.md for filed issues AUDIT.md collects the full audit (tests, lint, mypy, source review, runtime behavior) and maps the 25 raw findings to the 13 consolidated issues filed in this repo. ROADMAP.md sequences those 13 issues into five waves with parallelism called out, so contributors know which issues can be picked up simultaneously and which depend on earlier work landing first. Co-Authored-By: Claude Opus 4.7 (1M context) --- AUDIT.md | 318 +++++++++++++++++++++++++++++++++++++++++++++++++++++ ROADMAP.md | 105 ++++++++++++++++++ 2 files changed, 423 insertions(+) create mode 100644 AUDIT.md create mode 100644 ROADMAP.md diff --git a/AUDIT.md b/AUDIT.md new file mode 100644 index 0000000..58b5904 --- /dev/null +++ b/AUDIT.md @@ -0,0 +1,318 @@ +# `recost` (Python) — Audit Findings + +Date: 2026-05-13 +Audit scope: full SDK — tests, lint, mypy, source code, docs vs. reality, runtime behavior. + +## State of the build + +| Check | Result | +|---|---| +| `pytest` | 129 / 129 passing (~43s) | +| `ruff check recost/` | 1 error (unused import) | +| `mypy recost/` (strict) | **35 errors across 6 files** — README claims strict-clean. | +| `python -c "from recost import init"` | OK | + +Functionally the SDK runs; the mypy posture is broken, there's a real thread-safety bug in the aggregator, fork-safety is absent (breaking prefork servers), and naming is fragmented (`Recost` / `ReCost` / `EcoAPI*` are all referenced somewhere). The 13 consolidated issues below collect 25 original findings into the smallest set of focused, independently-fixable PRs. + +Priorities: **P0** = correctness / shipping bug, **P1** = should fix before next release, **P2** = polish. + +--- + +## Consolidated GitHub issues to file + +--- + +### 1. `Aggregator` is not thread-safe — flush vs. ingest race — P0 + +**Body:** + +`recost/_aggregator.py:127-132`'s `flush()` iterates `self._buckets.values()` while user threads concurrently call `ingest()` (via interceptor wrappers, which run on whatever thread issued the HTTP request). The flush is driven by a background `threading.Timer`. Concurrent dict mutation during iteration raises `RuntimeError: dictionary changed size during iteration`. + +Real race under any non-trivial load. There is no lock anywhere in `Aggregator`. + +**Fix:** wrap `ingest`, `flush`, and `would_overflow` with a `threading.RLock`. Add a regression test that pumps N worker threads into `ingest` while the timer thread calls `flush`. + +**Files:** `recost/_aggregator.py`, `tests/test_aggregator.py` + +**Includes:** original #1. + +--- + +### 2. 35 mypy strict errors — README claims strict-clean — P0 + +**Body:** + +`README.md` and `CLAUDE.md` both advertise `mypy --strict` cleanliness. `mypy recost/` reports 35 errors across 6 files. Highlights: `_interceptor.py`'s `latency_ms: int` parameter receives a `float`; `frameworks/flask.py:33`'s `self._handle = None` is inferred as `None` and breaks reassignment; stale `# type: ignore` comments flagged as unused. + +**Fix:** decide on the actual mypy contract. Either run mypy in CI and fix the 35 errors, or drop the strict-clean claim from the README. The first option is correct. + +**Files:** `recost/_interceptor.py`, `recost/frameworks/flask.py`, others; `README.md`; `pyproject.toml`; CI workflow. + +**Includes:** original #2. + +--- + +### 3. No fork-safety — `gunicorn` / `Celery` prefork workers report nothing — P0 + +**Body:** + +`recost/_init.py` registers no `os.register_at_fork` hooks. After a fork: +- Patched method-object references are inherited (fine — class attrs). +- The module-level `_handle` global points at the parent's `RecostHandle`, but **no timer thread is running in the child**. Flushes never fire. +- `Transport._local._loop` references the parent's asyncio loop, which doesn't exist in the child. `run_coroutine_threadsafe` raises and is swallowed (`_transport.py:195-197`). All events queue silently and never send. + +Symptom: every gunicorn pre-fork or Celery worker started after `init()` collects metrics into the void. Most production Python deployments use prefork servers — this is the single largest production gap. + +**Fix:** register `os.register_at_fork(after_in_child=_reinit_after_fork)` that re-creates the timer thread and transport thread in the child. Alternatively, document a "call `init()` in the worker's `post_fork` hook" pattern and refuse to instrument until `init()` has been called in the current PID. + +**Files:** `recost/_init.py`, `recost/_transport.py`, `README.md`, `tests/test_init.py` + +**Includes:** original #16. + +--- + +### 4. Module-level state races: `_handle`, `install`/`uninstall`, init-vs-dispose — P1 + +**Body:** + +Multiple module globals are read/written without locks: + +1. `_init.py:78, 225` reads/writes `_handle` without a lock. Two threads racing `init()` can both pass the `if _handle is not None` guard, both call `install()`, and orphan the first thread's transport + aggregator (the background timer thread leaks). + +2. Independent of (1), the *patches and the callback registration* can desynchronize. Thread A in `dispose()` is mid-`_unpatch_*` while Thread B in `init()` calls `install(on_event)` and sees `_installed=True` (set by A but not yet cleared). B short-circuits — the new init has no patches. + +**Fix:** guard `init()`, `dispose()`, `install()`, and `uninstall()` with a single module-level `threading.RLock`. Add tests that race both pairs. + +**Files:** `recost/_init.py`, `recost/_interceptor.py`, `tests/test_init.py` + +**Includes:** original #3, #20. + +--- + +### 5. Naming chaos and stale docs across README and CLAUDE.md — P1 + +**Body:** + +Brand name and class names are used inconsistently: + +| Where | Spelling | +|---|---| +| `README.md:1, 3` | `Recost` | +| `README.md` Flask section + license | `ReCost` | +| `recost/__init__.py:1-5` docstring | `ReCost` | +| `recost/frameworks/flask.py:22` (class) | `ReCost` | +| `recost/frameworks/fastapi.py` (class) | `RecostMiddleware` | +| `CLAUDE.md:19, 27` | `EcoAPIHandle`, `EcoAPIConfig`, `EcoAPIMiddleware` — **none exist in code** | + +The FastAPI middleware is `RecostMiddleware`, the Flask extension is `ReCost` — different conventions for analogous classes. + +Stale doc claims piled on top: +- "21+ built-in rules" in both docs; actual count is **34 rules / 14 providers** (`_provider_registry.py:33` comment already says so). +- `README.md:106` documents `flush_interval: float (30.0)` as a top-level option; in code, `flush_interval` is the **deprecated** seconds option that emits a `DeprecationWarning`. The real option is `flush_interval_ms: int = 30_000`. Users who follow the README pull warnings into their logs. +- `flush_interval_ms`, `max_buckets`, `shutdown_flush_timeout_ms` are missing from the README config table. + +**Fix:** +1. Pick one brand spelling (recommend `Recost`). +2. Rename `ReCost` (Flask) → `RecostExtension` (or `Recost`), with a deprecation alias for one release. +3. Strip every `EcoAPI*` reference from `CLAUDE.md`. +4. Walk the README: update provider count, swap to `flush_interval_ms` as the documented option (mark `flush_interval` deprecated), add the missing config fields. +5. Add a test that asserts `len(BUILTIN_PROVIDERS) == 34` so future drift is caught. + +**Files:** `recost/frameworks/flask.py`, `recost/__init__.py`, `README.md`, `CLAUDE.md`, `tests/test_provider_registry.py` + +**Includes:** original #4, #5, #6. + +--- + +### 6. Process lifecycle: no `atexit` flush, no signal handlers, dispose can leak threads — P1 + +**Body:** + +Two related lifecycle gaps: + +1. **No `atexit` / signal handler.** The flush timer is a daemon thread — when the process exits normally (`sys.exit()`, end of `__main__`, AWS Lambda invocation completes, SIGTERM in a container), the daemon thread is killed and the current aggregator bucket is lost. Cron jobs, Lambdas, batch scripts, and one-shot CLIs report nothing unless the user manually calls `handle.dispose()`. + +2. **Dispose during connect leaks FDs.** `recost/_transport.py:133-146, 199-210`'s `dispose()` sets `self._running = False` and queues a sentinel. If the loop is currently inside `websockets.connect(url)` (blocking await on TCP connect), the sentinel sits in the queue until the connect either succeeds or hits the OS TCP timeout (~75s on Linux). `thread.join(timeout=2.0)` returns without joining. The daemon thread + open socket FD leak until process exit. In long-lived processes that call `init()/dispose()` many times (test suites, Flask dev server with reload), FDs accumulate. + +**Fix:** +- In `init()`, register `atexit.register(_final_flush)` and (optionally) `signal.signal(SIGTERM, ...)`. Make handlers idempotent. Respect `shutdown_flush_timeout_ms`. Provide `auto_shutdown_handlers=False` opt-out. +- On dispose, `loop.call_soon_threadsafe(loop.stop)` and cancel pending tasks. Then join with a timeout derived from `shutdown_flush_timeout_ms` (currently hardcoded 5s) and log if the thread didn't exit. + +**Files:** `recost/_init.py`, `recost/_transport.py`, `tests/test_init.py` + +**Includes:** original #9 (hardcoded join), #18 (no atexit), #19 (dispose leak). + +--- + +### 7. Local-mode WebSocket: unbounded queue, infinite reconnect, no auth handshake — P1 + +**Body:** + +Three related local-transport hardening issues: + +1. **Queue unbounded.** `_transport.py:194` enqueues outbound messages with no cap. The re-queue path on disconnect (`_transport.py:176`) also uses `put_nowait`. A long extension outage = unbounded memory growth. + +2. **Reconnect loops forever.** `_transport.py:148-184` retries `localhost:9847` forever (capped at 30s per attempt). A user who deploys with no `api_key` (default = local mode) and no VS Code extension running keeps a daemon thread retrying forever while events queue up. Typical misconfig; silent failure. + +3. **No auth on the WS port.** The SDK sends serialized `WindowSummary` payloads to whatever process responds first on `127.0.0.1:9847`. Any local process can squat the port and silently sink all telemetry. Low risk on a dev machine, but easy to harden. + +**Fix:** +- Cap the queue (e.g. `asyncio.Queue(maxsize=1000)`) with drop-oldest. Log a single warning on first drop; reset on reconnect. +- After N consecutive failed connects (e.g. 10), give up and emit `on_error` once. Detect "no `api_key` AND first connect failed" and warn loudly. +- Lightweight handshake: SDK opens, sends `{"type":"hello","sdk":"recost-py","version":...}`; extension replies `{"type":"ack"}`. On no-ack within N ms, drop the connection without sending payloads. Coordinated change in the VS Code extension repo. + +**Files:** `recost/_transport.py`, `tests/test_transport.py`, plus a paired change in the extension. + +**Includes:** original #8, #17, #24. + +--- + +### 8. Interceptor body-size measurement is wrong for common patterns — P1 + +**Body:** + +Three related body-sizing issues: + +1. **aiohttp `json=` and `FormData` report 0 bytes.** `recost/_interceptor.py:317-325` checks only the `data=` kwarg. The most common aiohttp POST patterns (`session.post(url, json={...})`, `session.post(url, data=FormData(...))`, async-iterable bodies, `BytesIO`) all fall through and report 0. + +2. **httpx streaming body silently materialized.** `_interceptor.py:198-201, 244-247` accesses `request.content` to compute body size. For ordinary requests built via client methods, `content` is bytes — fine. For users passing a custom streaming body (`httpx.Request("POST", url, content=async_iterator)`), accessing `content` reads and buffers the entire iterator. A large upload silently OOMs the process. + +3. **Response body size derived from `Content-Length` only.** Chunked / streaming responses (LLM SSE streams) don't set the header; they always report `response_bytes=0`. The README's "response body size (bytes)" promise is partially false for streams. + +**Fix:** +- For aiohttp: when `json=` is present, JSON-serialize and measure; for `FormData`, query its `_size`; for unknown body types, leave at 0 but document. +- For httpx: `isinstance(request.content, bytes)` check before reading. Non-bytes → skip size measurement. +- For responses: document the streaming caveat in README; optionally tee non-streaming response bodies. + +**Files:** `recost/_interceptor.py`, `README.md`, `tests/test_interceptor.py` + +**Includes:** original #15 (Content-Length), #21 (aiohttp json), #22 (httpx streaming). + +--- + +### 9. Test gaps: aiohttp paths, privacy claim, self-instrumentation, 5xx retry — P1 + +**Body:** + +Four explicit claims in the README have no test coverage: + +- **aiohttp interceptor branch.** `_interceptor.py` patches `aiohttp.ClientSession._request`, but `tests/test_interceptor.py` only covers urllib3, httpx sync, and httpx async. Zero direct aiohttp tests. +- **No headers/bodies captured (privacy contract).** Add a test that asserts `RawEvent`-shaped output carries no dict-typed payload field, plus a test that issues a request with sensitive headers and confirms they don't surface. +- **No self-instrumentation.** Verify that `urllib.request` calls made from `_post_cloud` do not trigger the urllib3 patch. +- **5xx retry path.** `test_no_retry_on_4xx` exists but no positive test that `max_retries` actually runs `n` attempts with exponential backoff. + +Also missing: deprecation-warning test for `flush_interval`, `would_overflow` early-flush path test, `handle.dispose()` actually stops new flushes, `_LocalTransport` graceful no-op when websockets missing, Flask graceful degradation when flask is missing. + +**Fix:** add tests for each. The aiohttp tests should mirror the existing httpx async tests (success, 4xx capture, latency, response-bytes, double-count guard via reentrancy). + +**Files:** `tests/test_interceptor.py`, `tests/test_transport.py`, `tests/test_init.py`, `tests/test_flask.py` + +**Includes:** original #7, #14. + +--- + +### 10. Flush loop hygiene: errors swallowed indefinitely without backoff — P2 + +**Body:** + +`recost/_init.py:194-195` catches every exception in the flush loop and continues firing every `flush_interval_ms`. A deterministic bug (malformed metric, transport never reachable) silently re-fires forever, logging every cycle. + +**Fix:** track consecutive failures; after N (e.g. 5), back off exponentially up to a ceiling, and surface via `on_error`. + +**Files:** `recost/_init.py`, `tests/test_init.py` + +**Includes:** original #10. + +--- + +### 11. urllib3 wrapper maintenance: dead kwargs + brittle import-time defaults — P2 + +**Body:** + +Two related urllib3-patch maintenance issues: + +1. `_interceptor.py:108` lists `pool_connections` and `pool_maxsize` as kwargs on the `urlopen` wrapper. Neither is an actual `HTTPConnectionPool.urlopen` parameter — they're `PoolManager.__init__` parameters. The wrapper accepts and silently drops them. + +2. `_interceptor.py:108` references `urllib3.util.Timeout.DEFAULT_TIMEOUT` as a **default-argument value at function-definition time**. If the installed urllib3 (or a future release) doesn't expose that attribute, the SDK fails to import — and the failure happens before any user code runs. + +**Fix:** +- Remove the dead `pool_connections` / `pool_maxsize` params; rely entirely on `**kwargs` forwarding. +- Use a sentinel (`_DEFAULT = object()`) for the timeout default; resolve to the urllib3 default inside the body. Wrap import-time lookups in a try/except so a broken urllib3 install degrades to a no-op rather than crashing the host app. + +**Files:** `recost/_interceptor.py`, `tests/test_interceptor.py` + +**Includes:** original #11, #23. + +--- + +### 12. `exclude_patterns` is unscoped substring; localhost not auto-excluded in cloud mode — P2 + +**Body:** + +`_init.py:148-150` uses `pattern in event.url or pattern in event.host`. Short or hostname-like patterns over-match. `*` is taken literally, not as a glob — users naturally pass `"*.internal.corp"` expecting it to work and silently miss every request. + +Also: when `api_key` is set (cloud mode), the SDK does not auto-exclude `localhost` / `127.0.0.1`. A local dev recost instance could be self-traced. + +**Fix:** +- Add an option for exact host match (e.g. accept `("=", "api.example.com")` tuples or a separate `exclude_hosts` field). +- Document the substring contract explicitly; reject patterns containing `*` with a clear error so users don't misuse it. +- Auto-exclude localhost when a local recost dev API is detected. + +**Files:** `recost/_init.py`, `README.md`, `tests/test_init.py` + +**Includes:** original #12. + +--- + +### 13. Code hygiene: unused import, traceback context dropped — P2 + +**Body:** + +Two trivial fixes: + +1. `ruff check` reports `F401 unused-import` for `MAX_BUCKETS` in `recost/_transport.py:23`. +2. `recost/_interceptor.py:148-156, 219-228, 266-275, 342-350` re-raise via `raise exc` rather than bare `raise`, rebinding the traceback. End users debugging SDK-wrapped errors see the SDK wrapper at the top of the stack, not their own call site. + +**Fix:** +- Remove the unused import. +- Replace every `raise exc` with bare `raise`. + +**Files:** `recost/_transport.py`, `recost/_interceptor.py` + +**Includes:** original #13, #25. + +--- + +## Filing checklist + +- [ ] Open issues 1–13 in the Python repo. +- [ ] Label by priority (`P0` / `P1` / `P2`) and kind (`bug` / `docs` / `test` / `runtime`). +- [ ] Group milestones: + - **Next patch release**: issues 1, 2, 3 — the three P0s. Issue 3 (fork-safety) is the single largest production gap; most Python deployments use prefork servers. + - **Next minor release**: issues 4, 5, 6, 7, 8, 9 — P1 cluster (state safety, docs, lifecycle, transport, body sizing, test gaps). + - **Backlog**: 10–13. +- [ ] Link 4, 6 — both touch the lifecycle / threading model; consider one consolidated PR. +- [ ] Link 5 (naming + docs) — pure documentation PR, easy to land first and unblock everything else. +- [ ] Link 7 — coordinated change with the VS Code extension repo for the handshake. + +--- + +## Mapping back to original findings + +This consolidation collapses 25 raw findings into 13 issues by grouping by fix location and shared rationale. Mapping: + +| Consolidated | Original | +|---|---| +| 1 | #1 (aggregator race) | +| 2 | #2 (mypy errors) | +| 3 | #16 (fork-safety) | +| 4 | #3 (`_handle` race), #20 (init/dispose race) | +| 5 | #4 (naming), #5 (`flush_interval` deprecation), #6 (provider count) | +| 6 | #9 (hardcoded timer join), #18 (no atexit), #19 (dispose-during-connect leak) | +| 7 | #8 (queue unbounded), #17 (infinite reconnect), #24 (no auth) | +| 8 | #15 (Content-Length), #21 (aiohttp `json=`), #22 (httpx streaming) | +| 9 | #7 (aiohttp tests), #14 (privacy/self/5xx) | +| 10 | #10 (flush-loop errors) | +| 11 | #11 (dead kwargs), #23 (Timeout import-time) | +| 12 | #12 (exclude pattern) | +| 13 | #13 (ruff F401), #25 (`raise exc`) | diff --git a/ROADMAP.md b/ROADMAP.md new file mode 100644 index 0000000..827210d --- /dev/null +++ b/ROADMAP.md @@ -0,0 +1,105 @@ +# Working Order — `recost` Python SDK + +Plan for working through the 13 GitHub issues from [`AUDIT.md`](./AUDIT.md). Issues are grouped into **waves**; each wave is a set that can be worked in parallel without conflict, and waves are sequential. + +Mapping to filed issues: + +--- + +## At-a-glance + +| Wave | Issues | Parallel? | Reason for order | +|---|---|---|---| +| 1 — Foundation | #1, #5 | Yes (different files) | Both zero-risk for blocking other work. | +| 2 — Threading primitive | #4 | — | Adds module-level `RLock` that Wave 3 / 4 depend on. | +| 3 — Lifecycle | #6 then #7 | Sequential (both touch `_transport.py`) | #6 fixes dispose semantics first; #7 builds on clean dispose. | +| 4 — Fork-safety + body sizing | #3, #8 | Yes (different files) | #3 builds on Wave 3's clean dispose; #8 is `_interceptor.py`-only. | +| 5 — P2 cleanup + types + tests | #10, #11, #12, #13, #2, #9 | Mostly yes | Small surface; fold tests into each fix; mypy sweep cleans up residual. | + +--- + +## Detailed sequencing + +### Wave 1 — Foundation + +Two independent tracks, fully parallel. + +- **#1 — [Aggregator thread-safety](https://github.com/recost-dev/middleware-python/issues/1)** (P0) + Files: `recost/_aggregator.py`, `tests/test_aggregator.py` + Wrap `ingest` / `flush` / `would_overflow` in `threading.RLock`. Add a regression test that races N threads ingesting against a flush. No dependencies on other issues. + +- **#5 — [Naming + docs reconciliation](https://github.com/recost-dev/middleware-python/issues/5)** (P1) + Files: `recost/frameworks/flask.py`, `recost/__init__.py`, `README.md`, `CLAUDE.md`, `tests/test_provider_registry.py` + Flask class rename (`ReCost` → `RecostExtension`, alias for one release), strip `EcoAPI*` from CLAUDE.md, fix the provider-count claims (21 → 34), document `flush_interval_ms`/`max_buckets`/`shutdown_flush_timeout_ms`, mark `flush_interval` deprecated. No business-logic risk. + +**Why this order:** both are zero-risk for blocking other work. Naming should land before any test or doc changes elsewhere so subsequent commits don't reference the old names. + +--- + +### Wave 2 — Threading primitive + +- **#4 — [Module-level state races](https://github.com/recost-dev/middleware-python/issues/4)** (P1) + Files: `recost/_init.py`, `recost/_interceptor.py`, `tests/test_init.py` + Introduces a module-level `threading.RLock` around `init()` / `dispose()` / `install()` / `uninstall()`. **Must land before Wave 3** because #6 and #3 will both rely on the same lock for their atomicity guarantees. + +--- + +### Wave 3 — Process lifecycle + +Both edits live in `_transport.py`, so do them sequentially to avoid merge conflicts: + +1. **#6 — [Process lifecycle: atexit + dispose-during-connect FD leak](https://github.com/recost-dev/middleware-python/issues/6)** (P1) + Files: `recost/_init.py`, `recost/_transport.py`, `tests/test_init.py` + Fixes the dispose path so the loop properly stops and the thread joins. Adds an `atexit` handler that runs a final flush. Provides the clean-dispose mechanic that #3 reuses. + +2. **#7 — [Local-mode WebSocket hardening](https://github.com/recost-dev/middleware-python/issues/7)** (P1) + Files: `recost/_transport.py`, `tests/test_transport.py`, plus a coordinated change in the extension repo + Cap the queue (drop-oldest), give up after N failed reconnects (with a one-shot `on_error`), add lightweight `hello`/`ack` handshake. Doing this second means rebasing on top of cleaner dispose semantics. + +--- + +### Wave 4 — Fork-safety + body sizing + +Parallel — different files: + +- **#3 — [No fork-safety](https://github.com/recost-dev/middleware-python/issues/3)** (P0) + Files: `recost/_init.py`, `recost/_transport.py`, `README.md`, `tests/test_init.py` + Register `os.register_at_fork(after_in_child=_reinit_after_fork)` that re-creates the timer thread and transport thread in the child. Builds on Wave 3's clean dispose because the child needs to fully discard the parent's transport state. + +- **#8 — [Body-size measurement](https://github.com/recost-dev/middleware-python/issues/8)** (P1) + Files: `recost/_interceptor.py`, `README.md`, `tests/test_interceptor.py` + aiohttp `json=`/`FormData`, httpx streaming-body materialization, response Content-Length-only caveat. Independent of all lifecycle work. + +--- + +### Wave 5 — P2 cleanup + types + tests + +Most can be done in parallel; finish in any order. + +- **#10 — [Flush loop hygiene](https://github.com/recost-dev/middleware-python/issues/10)** (P2) — `recost/_init.py` +- **#11 — [urllib3 wrapper maintenance](https://github.com/recost-dev/middleware-python/issues/11)** (P2) — `recost/_interceptor.py` +- **#12 — [exclude_patterns](https://github.com/recost-dev/middleware-python/issues/12)** (P2) — `recost/_init.py`, `README.md` +- **#13 — [Code hygiene](https://github.com/recost-dev/middleware-python/issues/13)** (P2) — `recost/_transport.py`, `recost/_interceptor.py` +- **#2 — [35 mypy strict errors](https://github.com/recost-dev/middleware-python/issues/2)** (P0) — touched files were largely fixed during earlier waves; this is the residual sweep. Add `mypy --strict` to CI as part of this PR so the strict-clean claim becomes verifiable. +- **#9 — [Test gaps](https://github.com/recost-dev/middleware-python/issues/9)** (P1) — fold tests into each PR above. Outstanding items by end of Wave 4: aiohttp interceptor branch coverage (Wave 4 dependency), privacy test, self-instrumentation test, 5xx retry test. + +--- + +## Cross-cutting rules + +- **One issue per PR.** Each PR closes one numbered issue. Avoid bundling unrelated fixes. +- **Tests with every fix.** Don't merge a bugfix without a test that fails before the fix and passes after. +- **mypy `--strict` must pass on touched files before merge.** This is how #2 closes — incrementally per file. By Wave 5 only residual errors should remain. +- **Update `CHANGELOG.md` with each PR** (or create one if absent). + +--- + +## What to start with right now + +Pick one of: + +1. **#5** — low-risk warm-up that unblocks naming. +2. **#1** — go straight to a P0 with a focused fix. +3. **#4** — set up the threading primitive that 3 / 6 / 7 depend on (longer chain). + +**Recommendation:** open two branches and do **#5 + #1 in parallel as Wave 1**, then start **#4**. From 15307600c979596a77888ab7348176e5c7a5f57a Mon Sep 17 00:00:00 2001 From: Andres Lopez <190146319+AndresL230@users.noreply.github.com> Date: Wed, 13 May 2026 02:55:46 -0400 Subject: [PATCH 4/7] Add Wave 1 implementation plans (#1 and #5) Two TDD-structured plans for the parallel Wave 1 work identified in ROADMAP.md: - 2026-05-13-aggregator-thread-safety.md walks through adding a threading.RLock and swap-and-process flush, with a stress test that reliably reproduces the race before the fix lands. Closes #1. - 2026-05-13-naming-and-docs-reconciliation.md walks through renaming the Flask extension class, stripping nonexistent EcoAPI* references from CLAUDE.md, swapping the deprecated flush_interval option for flush_interval_ms in the README, documenting max_buckets and shutdown_flush_timeout_ms, and pinning the BUILTIN_PROVIDERS count. Closes #5. Both plans contain bite-sized tasks with the exact code/diffs to apply and the exact pytest commands to verify each step. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../2026-05-13-aggregator-thread-safety.md | 585 ++++++++++++++ ...26-05-13-naming-and-docs-reconciliation.md | 731 ++++++++++++++++++ 2 files changed, 1316 insertions(+) create mode 100644 docs/superpowers/plans/2026-05-13-aggregator-thread-safety.md create mode 100644 docs/superpowers/plans/2026-05-13-naming-and-docs-reconciliation.md diff --git a/docs/superpowers/plans/2026-05-13-aggregator-thread-safety.md b/docs/superpowers/plans/2026-05-13-aggregator-thread-safety.md new file mode 100644 index 0000000..3ffdfae --- /dev/null +++ b/docs/superpowers/plans/2026-05-13-aggregator-thread-safety.md @@ -0,0 +1,585 @@ +# Aggregator Thread-Safety Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make `recost._aggregator.Aggregator` safe to use concurrently from the interceptor's user threads (calling `ingest`) and the background timer thread (calling `flush`), eliminating the `RuntimeError: dictionary changed size during iteration` race documented in [issue #1](https://github.com/recost-dev/middleware-python/issues/1). + +**Architecture:** Add a single `threading.RLock` on `Aggregator`. Use a swap-and-process pattern in `flush`: under the lock, swap the buckets dict for a fresh empty one and capture window state; release the lock; then run the (potentially slow) percentile + summary construction work outside the critical section. Lock `ingest`, `would_overflow`, and the `size` / `bucket_count` properties for consistent snapshots. Add a stress test that races N ingester threads against a flusher thread and a correctness test that proves no events are lost. + +**Tech Stack:** Python ≥ 3.9 stdlib (`threading.RLock`, `concurrent.futures`), pytest, no new dependencies. + +--- + +## File Structure + +| File | Role | +|---|---| +| `recost/_aggregator.py` | Add `self._lock` to `__init__`; wrap mutating methods; refactor `flush` to swap-and-process. | +| `tests/test_aggregator.py` | Add `TestThreadSafety` class with two tests (no-exception stress test, no-lost-events correctness test). | + +Nothing else is touched. The change is fully isolated — no signature changes, no public API churn. + +--- + +## Task 1: Add the failing thread-safety stress test + +**Files:** +- Test: `tests/test_aggregator.py` (append a new `TestThreadSafety` class at the bottom) + +- [ ] **Step 1: Write the failing stress test** + +Append to `tests/test_aggregator.py` (after `class TestBucketOverflow` and before EOF): + +```python +# --------------------------------------------------------------------------- +# Thread safety +# --------------------------------------------------------------------------- + +import threading + + +class TestThreadSafety: + """Regression tests for issue #1 — Aggregator must be safe under concurrent + ingest (user threads) + flush (timer thread).""" + + def test_concurrent_ingest_and_flush_does_not_raise(self): + """Stress: 4 ingester threads run 5000 ingests each (20k total) while + one flusher thread runs 100 flushes. Pre-fix this reliably raises + ``RuntimeError: dictionary changed size during iteration``. + Post-fix it must complete without any exception.""" + agg = Aggregator() + exceptions: list[BaseException] = [] + iterations_per_ingester = 5000 + num_ingesters = 4 + num_flushes = 100 + + def ingester() -> None: + try: + for i in range(iterations_per_ingester): + p = f"p{i % 20}" + agg.ingest(make_event(provider=p, endpoint_category=p)) + except BaseException as exc: # noqa: BLE001 — we want everything + exceptions.append(exc) + + def flusher() -> None: + try: + for _ in range(num_flushes): + agg.flush() + except BaseException as exc: # noqa: BLE001 + exceptions.append(exc) + + threads = [threading.Thread(target=ingester) for _ in range(num_ingesters)] + threads.append(threading.Thread(target=flusher)) + for t in threads: + t.start() + for t in threads: + t.join(timeout=15.0) + + assert not exceptions, f"unexpected exceptions: {exceptions!r}" + # Sanity: at least some threads ran to completion. + for t in threads: + assert not t.is_alive(), "a worker thread is still running" +``` + +- [ ] **Step 2: Run the test and confirm it fails on the current code** + +Run: `python -m pytest tests/test_aggregator.py::TestThreadSafety::test_concurrent_ingest_and_flush_does_not_raise -v` + +Expected: **FAIL** with one of: +- `AssertionError: unexpected exceptions: [RuntimeError('dictionary changed size during iteration'), ...]` +- Or, on some Python builds, the lost-update side of the race may surface as a different exception. + +If the test passes pre-fix (rare but possible on a fast / cold machine), bump `iterations_per_ingester` to `20000` and re-run. It must reliably fail before continuing. + +- [ ] **Step 3: Do NOT commit yet** + +The test goes in the same commit as the fix (Task 2) so the repo never has a known-failing test on `main`. + +--- + +## Task 2: Add the lock and refactor `flush` to swap-and-process + +**Files:** +- Modify: `recost/_aggregator.py` + +- [ ] **Step 1: Add `threading` import and `self._lock` to `__init__`** + +Edit `recost/_aggregator.py`. Add `import threading` at the top with the other imports, and add the lock in `__init__`. + +Find: + +```python +from __future__ import annotations + +import math +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Dict, List, Optional + +from ._types import MetricEntry, RawEvent, WindowSummary +``` + +Replace with: + +```python +from __future__ import annotations + +import math +import threading +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Dict, List, Optional + +from ._types import MetricEntry, RawEvent, WindowSummary +``` + +Find: + +```python + def __init__( + self, + project_id: str = "", + environment: str = "development", + sdk_version: str = "0.0.0", + max_buckets: int = MAX_BUCKETS, + ) -> None: + self._project_id = project_id + self._environment = environment + self._sdk_version = sdk_version + self._max_buckets = max_buckets + self._buckets: Dict[str, _Bucket] = {} + self._window_start: Optional[str] = None + self._size = 0 +``` + +Replace with: + +```python + def __init__( + self, + project_id: str = "", + environment: str = "development", + sdk_version: str = "0.0.0", + max_buckets: int = MAX_BUCKETS, + ) -> None: + self._project_id = project_id + self._environment = environment + self._sdk_version = sdk_version + self._max_buckets = max_buckets + self._buckets: Dict[str, _Bucket] = {} + self._window_start: Optional[str] = None + self._size = 0 + # RLock so a single thread can re-enter (e.g. if a future event + # callback ever calls back into the aggregator). Guards every method + # below that reads or writes _buckets / _window_start / _size. + self._lock = threading.RLock() +``` + +- [ ] **Step 2: Wrap `would_overflow` in the lock** + +Find: + +```python + def would_overflow(self, event: RawEvent) -> bool: + """True if ingesting ``event`` would allocate a new bucket while the + window is already at capacity. Callers should flush before ingesting.""" + if len(self._buckets) < self._max_buckets: + return False + return self._key_for(event) not in self._buckets +``` + +Replace with: + +```python + def would_overflow(self, event: RawEvent) -> bool: + """True if ingesting ``event`` would allocate a new bucket while the + window is already at capacity. Callers should flush before ingesting.""" + with self._lock: + if len(self._buckets) < self._max_buckets: + return False + return self._key_for(event) not in self._buckets +``` + +- [ ] **Step 3: Wrap `ingest` in the lock** + +Find: + +```python + def ingest(self, event: RawEvent, cost_cents: float = 0.0) -> None: + """Add one RawEvent to the current window.""" + if self._window_start is None: + self._window_start = event.timestamp + + provider = event.provider if event.provider is not None else "unknown" + endpoint = event.endpoint_category if event.endpoint_category is not None else event.path + key = self._key_for(event) + + bucket = self._buckets.get(key) + if bucket is None: + bucket = _Bucket(provider=provider, endpoint=endpoint, method=event.method) + self._buckets[key] = bucket + + bucket.request_count += 1 + if event.error: + bucket.error_count += 1 + bucket.latencies.append(event.latency_ms) + bucket.total_request_bytes += event.request_bytes + bucket.total_response_bytes += event.response_bytes + bucket.estimated_cost_cents += cost_cents + + self._size += 1 +``` + +Replace with: + +```python + def ingest(self, event: RawEvent, cost_cents: float = 0.0) -> None: + """Add one RawEvent to the current window.""" + with self._lock: + if self._window_start is None: + self._window_start = event.timestamp + + provider = event.provider if event.provider is not None else "unknown" + endpoint = event.endpoint_category if event.endpoint_category is not None else event.path + key = self._key_for(event) + + bucket = self._buckets.get(key) + if bucket is None: + bucket = _Bucket(provider=provider, endpoint=endpoint, method=event.method) + self._buckets[key] = bucket + + bucket.request_count += 1 + if event.error: + bucket.error_count += 1 + bucket.latencies.append(event.latency_ms) + bucket.total_request_bytes += event.request_bytes + bucket.total_response_bytes += event.response_bytes + bucket.estimated_cost_cents += cost_cents + + self._size += 1 +``` + +- [ ] **Step 4: Refactor `flush` to swap-and-process** + +The current `flush` holds state while building the summary — that means sorting latencies for thousands of buckets runs inside the critical section. Swap the buckets dict atomically, then process the snapshot without the lock. + +Find: + +```python + def flush(self) -> Optional[WindowSummary]: + """Compress the current window into a WindowSummary and reset state.""" + if not self._buckets: + return None + + window_start = self._window_start or datetime.now(timezone.utc).isoformat() + window_end = datetime.now(timezone.utc).isoformat() + + metrics: List[MetricEntry] = [] + + for bucket in self._buckets.values(): + sorted_latencies = sorted(bucket.latencies) + total_latency_ms = sum(sorted_latencies) + + metrics.append(MetricEntry( + provider=bucket.provider, + endpoint=bucket.endpoint, + method=bucket.method, + request_count=bucket.request_count, + error_count=bucket.error_count, + total_latency_ms=total_latency_ms, + p50_latency_ms=_compute_percentile(sorted_latencies, 0.5), + p95_latency_ms=_compute_percentile(sorted_latencies, 0.95), + total_request_bytes=bucket.total_request_bytes, + total_response_bytes=bucket.total_response_bytes, + estimated_cost_cents=bucket.estimated_cost_cents, + )) + + # Reset + self._buckets = {} + self._window_start = None + self._size = 0 + + return WindowSummary( + project_id=self._project_id, + environment=self._environment, + sdk_language="python", + sdk_version=self._sdk_version, + window_start=window_start, + window_end=window_end, + metrics=metrics, + ) +``` + +Replace with: + +```python + def flush(self) -> Optional[WindowSummary]: + """Compress the current window into a WindowSummary and reset state. + + Swap-and-process: under the lock, take ownership of the current + buckets dict and reset state; then sort/percentile-compute outside + the lock so ingest is not blocked for the duration of the flush. + """ + with self._lock: + if not self._buckets: + return None + buckets_to_flush = self._buckets + window_start_captured = self._window_start + # Reset state before releasing the lock so concurrent ingests + # land in the *next* window, not this one. + self._buckets = {} + self._window_start = None + self._size = 0 + + # Outside the lock: format timestamps and build the summary. + window_end = datetime.now(timezone.utc).isoformat() + window_start = window_start_captured or window_end + + metrics: List[MetricEntry] = [] + for bucket in buckets_to_flush.values(): + sorted_latencies = sorted(bucket.latencies) + total_latency_ms = sum(sorted_latencies) + + metrics.append(MetricEntry( + provider=bucket.provider, + endpoint=bucket.endpoint, + method=bucket.method, + request_count=bucket.request_count, + error_count=bucket.error_count, + total_latency_ms=total_latency_ms, + p50_latency_ms=_compute_percentile(sorted_latencies, 0.5), + p95_latency_ms=_compute_percentile(sorted_latencies, 0.95), + total_request_bytes=bucket.total_request_bytes, + total_response_bytes=bucket.total_response_bytes, + estimated_cost_cents=bucket.estimated_cost_cents, + )) + + return WindowSummary( + project_id=self._project_id, + environment=self._environment, + sdk_language="python", + sdk_version=self._sdk_version, + window_start=window_start, + window_end=window_end, + metrics=metrics, + ) +``` + +- [ ] **Step 5: Wrap `size` and `bucket_count` properties in the lock** + +Find: + +```python + @property + def size(self) -> int: + """Total events ingested since the last flush.""" + return self._size + + @property + def bucket_count(self) -> int: + """Number of unique provider + endpoint + method groups.""" + return len(self._buckets) +``` + +Replace with: + +```python + @property + def size(self) -> int: + """Total events ingested since the last flush.""" + with self._lock: + return self._size + + @property + def bucket_count(self) -> int: + """Number of unique provider + endpoint + method groups.""" + with self._lock: + return len(self._buckets) +``` + +(`max_buckets` is configured at construction and is immutable for the lifetime of the instance — no lock needed.) + +- [ ] **Step 6: Run the stress test from Task 1 and confirm it passes** + +Run: `python -m pytest tests/test_aggregator.py::TestThreadSafety::test_concurrent_ingest_and_flush_does_not_raise -v` + +Expected: **PASS** within ~5 seconds. + +- [ ] **Step 7: Run the full aggregator test suite to confirm no regressions** + +Run: `python -m pytest tests/test_aggregator.py -v` + +Expected: every existing test still passes (the original suite has ~40 cases — count varies; all must remain green). + +- [ ] **Step 8: Commit the test + fix together** + +```bash +git add recost/_aggregator.py tests/test_aggregator.py +git commit -m "fix(aggregator): make Aggregator thread-safe with RLock + +Guards _buckets / _window_start / _size with a threading.RLock so the +background timer thread's flush() cannot race the interceptor's user +threads ingesting events. flush() now uses a swap-and-process pattern +so percentile computation runs outside the lock. + +Adds a stress-test regression that reliably reproduced the race on the +old code (RuntimeError: dictionary changed size during iteration). + +Closes #1" +``` + +--- + +## Task 3: Add a correctness test that proves no events are lost + +The stress test above only proves \"no exception is raised\". This task adds a second test that proves the lock also fixes the lost-update race on counter increments (`bucket.request_count += 1` is *not* atomic across threads even with a dict-level lock — but `with self._lock` around the whole `ingest` body covers it). + +**Files:** +- Test: `tests/test_aggregator.py` (extend `TestThreadSafety`) + +- [ ] **Step 1: Append the correctness test** + +Inside the existing `class TestThreadSafety:` (after `test_concurrent_ingest_and_flush_does_not_raise`): + +```python + def test_concurrent_correctness_no_lost_events(self): + """4 ingester threads each push 2000 events while a flusher pulls + windows concurrently. The total request_count summed across every + flushed summary, plus the request_count of the final drain, must + equal the total number of events ingested. Pre-fix this fails + because ``bucket.request_count += 1`` races across threads.""" + agg = Aggregator() + iterations_per_ingester = 2000 + num_ingesters = 4 + total_expected = iterations_per_ingester * num_ingesters + flushed_total = 0 + flushed_lock = threading.Lock() # this lock is in the *test*, not the SUT + exceptions: list[BaseException] = [] + ingest_done = threading.Event() + + def ingester() -> None: + try: + for i in range(iterations_per_ingester): + # Use a small key space so several events land in the same + # bucket and the counter increment is contested. + p = f"p{i % 3}" + agg.ingest(make_event(provider=p, endpoint_category=p)) + except BaseException as exc: # noqa: BLE001 + exceptions.append(exc) + + def flusher() -> None: + nonlocal flushed_total + try: + while not ingest_done.is_set(): + summary = agg.flush() + if summary is not None: + n = sum(m.request_count for m in summary.metrics) + with flushed_lock: + flushed_total += n + except BaseException as exc: # noqa: BLE001 + exceptions.append(exc) + + ingesters = [threading.Thread(target=ingester) for _ in range(num_ingesters)] + flusher_thread = threading.Thread(target=flusher) + flusher_thread.start() + for t in ingesters: + t.start() + for t in ingesters: + t.join(timeout=15.0) + ingest_done.set() + flusher_thread.join(timeout=5.0) + + # Final drain — anything ingested after the last flusher iteration. + final = agg.flush() + if final is not None: + flushed_total += sum(m.request_count for m in final.metrics) + + assert not exceptions, f"unexpected exceptions: {exceptions!r}" + assert flushed_total == total_expected, ( + f"lost events: flushed {flushed_total}, expected {total_expected}" + ) +``` + +- [ ] **Step 2: Run the correctness test and confirm it passes** + +Run: `python -m pytest tests/test_aggregator.py::TestThreadSafety::test_concurrent_correctness_no_lost_events -v` + +Expected: **PASS**. + +If it fails (`lost events: flushed N, expected M`), it means the lock isn't wrapping the full `ingest` body. Re-check Task 2 Step 3. + +- [ ] **Step 3: Commit** + +```bash +git add tests/test_aggregator.py +git commit -m "test(aggregator): assert concurrent ingest does not lose events + +Adds a second thread-safety test that proves no counter increments are +lost when N threads race on a shared bucket. Complements the no-raise +stress test by exercising the counter-increment side of the race." +``` + +--- + +## Task 4: Final verification — lint, types, full suite + +- [ ] **Step 1: Run the full test suite** + +Run: `python -m pytest` + +Expected: every test passes (including the 130-ish existing tests and the 2 new ones). + +- [ ] **Step 2: Run ruff** + +Run: `python -m ruff check recost/ tests/` + +Expected: no new errors. (Pre-existing `F401 MAX_BUCKETS` in `_transport.py` is a separate issue, [#13](https://github.com/recost-dev/middleware-python/issues/13), and is out of scope.) + +- [ ] **Step 3: Run mypy on the changed file** + +Run: `python -m mypy recost/_aggregator.py` + +Expected: no new errors introduced by this change. (The repo overall has 35 known mypy errors tracked under [#2](https://github.com/recost-dev/middleware-python/issues/2); this PR must not increase that count for `_aggregator.py`.) + +- [ ] **Step 4: Open the PR** + +```bash +git push -u origin +gh pr create --title "fix(aggregator): make Aggregator thread-safe (closes #1)" --body "$(cat <<'EOF' +## Summary + +Closes #1. + +Guards `Aggregator` state with a `threading.RLock` so the background timer +thread's `flush()` cannot race user threads calling `ingest()` via the +interceptor. `flush()` uses a swap-and-process pattern so percentile +computation runs outside the critical section. + +## Tests + +- New `TestThreadSafety::test_concurrent_ingest_and_flush_does_not_raise` — + 4 ingester threads × 5000 events vs. 100 flushes; reliably reproduced + the `RuntimeError: dictionary changed size during iteration` on the + old code. +- New `TestThreadSafety::test_concurrent_correctness_no_lost_events` — + proves counter increments are not lost across threads. + +## Notes + +- Public API unchanged. +- No new dependencies. +- Out of scope: thread-safety of `Aggregator.size` callers in `_init.py` — + the property is now snapshot-consistent, but the calling pattern itself + is part of issue #4. +EOF +)" +``` + +--- + +## Self-review + +- **Spec coverage:** The filed issue body requires (a) RLock around `ingest`/`flush`/`would_overflow` — done in Task 2 steps 2–4; (b) a regression test pumping N threads — done in Task 1 and Task 3. +- **Placeholder scan:** No TBDs. Every code step shows the exact diff or test body. +- **Type consistency:** `Dict[str, _Bucket]`, `List[MetricEntry]`, `Optional[str]`, `Optional[WindowSummary]` all match the existing module's type style; `list[BaseException]` in the test files is Python 3.9-compatible because the test module already uses `from __future__ import annotations` implicitly via no annotations — re-check at execution time and switch to `List[BaseException]` if mypy complains under `--strict`. +- **Dependencies on other issues:** None. This is the first wave-1 item; the worktree is created from `main`. diff --git a/docs/superpowers/plans/2026-05-13-naming-and-docs-reconciliation.md b/docs/superpowers/plans/2026-05-13-naming-and-docs-reconciliation.md new file mode 100644 index 0000000..e325667 --- /dev/null +++ b/docs/superpowers/plans/2026-05-13-naming-and-docs-reconciliation.md @@ -0,0 +1,731 @@ +# Naming and Docs Reconciliation Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Unify naming and documentation across the Python SDK so it has a single brand spelling (`Recost`), a single Flask-extension class name (`RecostExtension`, with a one-release `ReCost` deprecation alias), and accurate docs (real provider count, real config fields, no references to nonexistent `EcoAPI*` types). Closes [issue #5](https://github.com/recost-dev/middleware-python/issues/5). + +**Architecture:** +- Rename the Flask extension class `ReCost` → `RecostExtension` in `recost/frameworks/flask.py`. Keep `ReCost = RecostExtension` as a deprecated alias that emits a `DeprecationWarning` on instantiation so external consumers don't break in this release. Document removal in a future major. +- Add `RecostExtension` to the public surface in `recost/__init__.py`. +- Add a regression test in `tests/test_provider_registry.py` asserting `len(BUILTIN_PROVIDERS) == 34` so future drift trips CI. +- Walk `CLAUDE.md` and remove every `EcoAPI*` reference (none exist in code). +- Walk `README.md`: rename `ReCost` → `RecostExtension` in the Flask example, replace the `flush_interval` row in the config table with `flush_interval_ms` (canonical) + a deprecated-`flush_interval` row, add rows for `max_buckets` and `shutdown_flush_timeout_ms`. + +**Tech Stack:** Python ≥ 3.9 stdlib (`warnings`), pytest, no new dependencies. + +--- + +## File Structure + +| File | Role | +|---|---| +| `recost/frameworks/flask.py` | Rename `ReCost` → `RecostExtension`; add deprecation alias. | +| `recost/__init__.py` | Re-export `RecostExtension` (and the deprecation alias). | +| `tests/test_flask.py` | Update existing imports to new name; add deprecation-warning test. | +| `tests/test_provider_registry.py` | Add `len(BUILTIN_PROVIDERS) == 34` assertion. | +| `CLAUDE.md` | Strip `EcoAPI*`, fix provider-count claims, fix file-purpose comments. | +| `README.md` | Update Flask example, replace config table rows. | + +No code in `_init.py`, `_interceptor.py`, `_transport.py`, `_aggregator.py`, `_types.py`, or `_provider_registry.py` changes. This is a docs-and-rename PR. + +--- + +## Task 1: Rename the Flask class with a deprecation alias (TDD) + +**Files:** +- Modify: `recost/frameworks/flask.py` +- Test: `tests/test_flask.py` + +- [ ] **Step 1: Write the failing test for the new name + deprecation alias** + +Replace the entire contents of `tests/test_flask.py` with: + +```python +""" +Tests for recost/frameworks/flask.py +""" + +import warnings + +import pytest + +from recost._interceptor import is_installed, uninstall +from recost._types import RecostConfig + + +class TestRecostExtension: + def test_extension_initializes_interceptor(self): + try: + from flask import Flask + from recost.frameworks.flask import RecostExtension + + app = Flask(__name__) + RecostExtension(app, config=RecostConfig(enabled=True)) + assert is_installed() + finally: + uninstall() + + def test_extension_init_app_pattern(self): + try: + from flask import Flask + from recost.frameworks.flask import RecostExtension + + ext = RecostExtension() + app = Flask(__name__) + ext.init_app(app, config=RecostConfig(enabled=True)) + assert is_installed() + finally: + uninstall() + + def test_extension_accepts_kwargs(self): + try: + from flask import Flask + from recost.frameworks.flask import RecostExtension + + app = Flask(__name__) + RecostExtension(app, enabled=True, debug=False) + assert is_installed() + finally: + uninstall() + + +class TestReCostDeprecationAlias: + """The old `ReCost` name must keep working for one release but emit a + DeprecationWarning so users migrate to `RecostExtension`.""" + + def test_old_name_still_constructs(self): + try: + from flask import Flask + from recost.frameworks.flask import ReCost + + app = Flask(__name__) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + ReCost(app, config=RecostConfig(enabled=True)) + assert is_installed() + # At least one DeprecationWarning was emitted naming the new class. + deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)] + assert deprecations, "expected a DeprecationWarning on ReCost(...)" + assert "RecostExtension" in str(deprecations[0].message) + finally: + uninstall() + + def test_old_name_is_subclass_or_alias_of_new(self): + """Importing the old name must yield the same class as the new name + so isinstance checks and existing type annotations keep working.""" + from recost.frameworks.flask import ReCost, RecostExtension + + # Either `ReCost is RecostExtension` (pure alias) or + # `issubclass(ReCost, RecostExtension)` (thin subclass with __init__ + # that emits the warning). Both shapes are acceptable. + assert ReCost is RecostExtension or issubclass(ReCost, RecostExtension) +``` + +- [ ] **Step 2: Run the new tests and confirm they fail** + +Run: `python -m pytest tests/test_flask.py -v` + +Expected: every `TestRecostExtension::*` test FAILS with `ImportError: cannot import name 'RecostExtension'`. The deprecation-alias tests fail at import for the same reason. + +- [ ] **Step 3: Rename the class and add the deprecation alias** + +Replace `recost/frameworks/flask.py` contents with: + +```python +""" +Flask extension adapter for recost. + +Usage: + from flask import Flask + from recost.frameworks.flask import RecostExtension + + app = Flask(__name__) + ext = RecostExtension(app, api_key="...", project_id="...") +""" + +from __future__ import annotations + +import warnings +from typing import Any, Optional + +from .._init import init +from .._types import RecostConfig + +try: + from flask import Flask + + class RecostExtension: + """Flask extension that initializes Recost telemetry.""" + + def __init__( + self, + app: Optional[Flask] = None, + config: Optional[RecostConfig] = None, + **kwargs: Any, + ) -> None: + self._handle = None + if app is not None: + self.init_app(app, config, **kwargs) + + def init_app( + self, + app: Flask, + config: Optional[RecostConfig] = None, + **kwargs: Any, + ) -> None: + if config is None: + config = RecostConfig(**kwargs) + self._handle = init(config) + + class ReCost(RecostExtension): + """Deprecated alias for :class:`RecostExtension`. + + Will be removed in a future release. Switch to + ``from recost.frameworks.flask import RecostExtension``. + """ + + def __init__(self, *args: Any, **kwargs: Any) -> None: + warnings.warn( + "recost.frameworks.flask.ReCost is deprecated and will be " + "removed in a future release; use RecostExtension instead.", + DeprecationWarning, + stacklevel=2, + ) + super().__init__(*args, **kwargs) + +except ImportError: + class RecostExtension: # type: ignore[no-redef] + """Stub — install 'flask' to use: pip install recost[flask]""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + raise ImportError( + "flask is required for the Recost extension. " + "Install it with: pip install recost[flask]" + ) + + class ReCost(RecostExtension): # type: ignore[no-redef] + """Stub deprecated alias — install 'flask' to use.""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + warnings.warn( + "recost.frameworks.flask.ReCost is deprecated; use " + "RecostExtension instead.", + DeprecationWarning, + stacklevel=2, + ) + super().__init__(*args, **kwargs) +``` + +- [ ] **Step 4: Run the Flask tests and confirm they pass** + +Run: `python -m pytest tests/test_flask.py -v` + +Expected: all 5 tests PASS. + +- [ ] **Step 5: Commit** + +```bash +git add recost/frameworks/flask.py tests/test_flask.py +git commit -m "feat(flask): rename ReCost to RecostExtension; deprecate old name + +The Flask extension class is renamed to RecostExtension to match the +naming convention of RecostMiddleware (FastAPI) and the rest of the +public surface. The old ReCost class remains importable as a thin +subclass that emits a DeprecationWarning on construction. + +Refs #5" +``` + +--- + +## Task 2: Expose `RecostExtension` on the public `recost` surface + +**Files:** +- Modify: `recost/__init__.py` + +- [ ] **Step 1: Add `RecostExtension` to the top-level re-exports** + +Replace `recost/__init__.py` contents with: + +```python +""" +recost — Python SDK for Recost. + +Tracks outbound HTTP API calls and reports cost, latency, and usage patterns +to the Recost dashboard or your local VS Code extension. +""" + +from ._types import ( + FlushStatus, + RecostConfig, + MetricEntry, + ProviderDef, + RawEvent, + TransportMode, + WindowSummary, +) +from ._init import RecostHandle, init +from ._provider_registry import BUILTIN_PROVIDERS, MatchResult, ProviderRegistry +from ._interceptor import install, uninstall, is_installed +from ._aggregator import Aggregator, MAX_BUCKETS + +__all__ = [ + "init", + "RecostHandle", + "RawEvent", + "MetricEntry", + "WindowSummary", + "ProviderDef", + "RecostConfig", + "TransportMode", + "FlushStatus", + "ProviderRegistry", + "BUILTIN_PROVIDERS", + "MatchResult", + "install", + "uninstall", + "is_installed", + "Aggregator", + "MAX_BUCKETS", +] +``` + +Note: framework adapters (`flask.RecostExtension`, `fastapi.RecostMiddleware`) are intentionally **not** re-exported on the top-level `recost` namespace because they have optional-dep imports. Users access them via `from recost.frameworks.flask import RecostExtension`. The only docstring change here is brand spelling. + +- [ ] **Step 2: Run the existing smoke tests to confirm imports still work** + +Run: `python -m pytest tests/test_scaffold.py -v` + +Expected: all PASS. + +Run: `python -c "from recost import init, RecostHandle, RecostConfig; print('ok')"` + +Expected: `ok`. + +- [ ] **Step 3: Commit** + +```bash +git add recost/__init__.py +git commit -m "docs(__init__): standardize on 'Recost' spelling in module docstring + +Refs #5" +``` + +--- + +## Task 3: Pin the BUILTIN_PROVIDERS count + +**Files:** +- Test: `tests/test_provider_registry.py` + +- [ ] **Step 1: Read the existing test file header to find the right spot** + +Run: `python -m pytest tests/test_provider_registry.py -v --collect-only` + +Expected: lists ~30+ test cases. Confirm `class TestBuiltinProviders` or similar exists (or add the new test as a free-standing function — either is fine). + +- [ ] **Step 2: Update the stale docstring and add the count assertion** + +Find this line in `tests/test_provider_registry.py:14`: + +```python + """Tests for all 21 built-in provider rules.""" +``` + +Replace with: + +```python + """Tests for all 34 built-in provider rules (14 unique providers).""" +``` + +Then append (at the bottom of the file, before any trailing newline / EOF): + +```python +# --------------------------------------------------------------------------- +# Built-in provider count — pins the published claim +# --------------------------------------------------------------------------- + +def test_builtin_providers_count_is_pinned(): + """If you add or remove a built-in provider rule, update this assertion + AND update the provider-count claims in README.md and CLAUDE.md.""" + from recost._provider_registry import BUILTIN_PROVIDERS + + assert len(BUILTIN_PROVIDERS) == 34, ( + f"BUILTIN_PROVIDERS has {len(BUILTIN_PROVIDERS)} rules; " + f"docs claim 34. Update docs and this assertion together." + ) +``` + +- [ ] **Step 3: Run the test** + +Run: `python -m pytest tests/test_provider_registry.py::test_builtin_providers_count_is_pinned -v` + +Expected: PASS (current count is 34). + +- [ ] **Step 4: Run the full provider-registry suite** + +Run: `python -m pytest tests/test_provider_registry.py -v` + +Expected: all tests pass (only the docstring changed in the existing class). + +- [ ] **Step 5: Commit** + +```bash +git add tests/test_provider_registry.py +git commit -m "test(provider-registry): pin BUILTIN_PROVIDERS count to 34 + +Adds a regression assertion so adding/removing built-in rules forces +a corresponding docs update. Also corrects the stale '21 built-in +provider rules' docstring. + +Refs #5" +``` + +--- + +## Task 4: Rewrite `CLAUDE.md` — strip `EcoAPI*`, fix provider count + +**Files:** +- Modify: `CLAUDE.md` + +`CLAUDE.md` has six concrete problems (verified by grep): + +| Line | Current | Why wrong | +|---|---|---| +| 19 | `# Main entry point — wires interceptor, registry, aggregator, transport; returns EcoAPIHandle` | `EcoAPIHandle` doesn't exist; it's `RecostHandle`. | +| 20 | `# All types: RawEvent, MetricEntry, WindowSummary, ProviderDef, EcoAPIConfig, TransportMode` | `EcoAPIConfig` doesn't exist; it's `RecostConfig`. Also missing `FlushStatus`. | +| 21 | `# ProviderRegistry — 21+ built-in rules, wildcard host matching, custom provider priority` | Actual count is 34. | +| 27 | `# EcoAPIMiddleware — ASGI middleware for FastAPI/Starlette` | Class is `RecostMiddleware`. | +| 28 | `# EcoAPI — Flask extension with init_app() pattern` | After Task 1, class is `RecostExtension` (with deprecated `ReCost` alias). | +| 33 | `# All 21 built-in providers, wildcards, Twilio refinement, custom priority` | Actual count is 34. | +| 66 | `21+ built-in rules covering:` | Actual count is 34. | + +- [ ] **Step 1: Apply six edits to `CLAUDE.md`** + +Edit 1 — line 19: + +Find: + +``` + _init.py # Main entry point — wires interceptor, registry, aggregator, transport; returns EcoAPIHandle +``` + +Replace with: + +``` + _init.py # Main entry point — wires interceptor, registry, aggregator, transport; returns RecostHandle +``` + +Edit 2 — line 20: + +Find: + +``` + _types.py # All types: RawEvent, MetricEntry, WindowSummary, ProviderDef, EcoAPIConfig, TransportMode +``` + +Replace with: + +``` + _types.py # All types: RawEvent, MetricEntry, WindowSummary, ProviderDef, RecostConfig, FlushStatus, TransportMode +``` + +Edit 3 — line 21: + +Find: + +``` + _provider_registry.py # ProviderRegistry — 21+ built-in rules, wildcard host matching, custom provider priority +``` + +Replace with: + +``` + _provider_registry.py # ProviderRegistry — 34 built-in rules (14 providers), wildcard host matching, custom provider priority +``` + +Edit 4 — line 27: + +Find: + +``` + fastapi.py # EcoAPIMiddleware — ASGI middleware for FastAPI/Starlette +``` + +Replace with: + +``` + fastapi.py # RecostMiddleware — ASGI middleware for FastAPI/Starlette +``` + +Edit 5 — line 28: + +Find: + +``` + flask.py # EcoAPI — Flask extension with init_app() pattern +``` + +Replace with: + +``` + flask.py # RecostExtension — Flask extension with init_app() pattern (ReCost is a deprecated alias) +``` + +Edit 6 — line 33: + +Find: + +``` + test_provider_registry.py # All 21 built-in providers, wildcards, Twilio refinement, custom priority +``` + +Replace with: + +``` + test_provider_registry.py # All 34 built-in provider rules, wildcards, Twilio refinement, custom priority +``` + +Edit 7 — line 66: + +Find: + +``` +21+ built-in rules covering: +``` + +Replace with: + +``` +34 built-in rules across 14 providers: +``` + +- [ ] **Step 2: Verify all stale references are gone** + +Run: `python -m grep -nE "EcoAPI|21\+? built-in|21 built-in" CLAUDE.md` (or use ripgrep / VS Code search). + +Expected: no matches. + +- [ ] **Step 3: Commit** + +```bash +git add CLAUDE.md +git commit -m "docs(CLAUDE): strip nonexistent EcoAPI* names; fix provider count + +CLAUDE.md referenced EcoAPIHandle, EcoAPIConfig, EcoAPIMiddleware, and +EcoAPI — none of which exist in the codebase. Replace with the actual +class names (RecostHandle, RecostConfig, RecostMiddleware, +RecostExtension). Correct the provider-rule count from '21+' to '34'. + +Refs #5" +``` + +--- + +## Task 5: Update `README.md` — Flask example, config table + +**Files:** +- Modify: `README.md` + +Three concrete problems in `README.md` (verified by grep + read): + +1. Lines 84, 87, 93 reference the `ReCost` class — update to `RecostExtension`. +2. Line 106 documents `flush_interval` (deprecated seconds option) as the canonical option — replace with `flush_interval_ms` and add a row for the deprecated form. +3. `max_buckets` and `shutdown_flush_timeout_ms` are not documented at all. + +- [ ] **Step 1: Update the Flask example** + +Find (in `README.md`, around lines 80–95): + +``` +### Flask + +```python +from flask import Flask +from recost.frameworks.flask import ReCost + +app = Flask(__name__) +ReCost(app, api_key="...", project_id="...") +``` + +Or using the `init_app` pattern: + +```python +recost = ReCost() +recost.init_app(app, api_key="...", project_id="...") +``` +``` + +Replace with: + +```` +### Flask + +```python +from flask import Flask +from recost.frameworks.flask import RecostExtension + +app = Flask(__name__) +RecostExtension(app, api_key="...", project_id="...") +``` + +Or using the `init_app` pattern: + +```python +ext = RecostExtension() +ext.init_app(app, api_key="...", project_id="...") +``` + +> **Note:** the old class name `ReCost` is still importable as a deprecated +> alias and will continue to work for one release with a `DeprecationWarning`. +> Migrate to `RecostExtension`. +```` + +- [ ] **Step 2: Update the config table** + +Find (in `README.md`, lines 101–115): + +``` +| Option | Type | Default | Description | +|---|---|---|---| +| `api_key` | `str` | — | Recost API key (`rc-...`). If omitted, runs in local mode. | +| `project_id` | `str` | — | Recost project ID. Required in cloud mode. | +| `environment` | `str` | `"development"` | Environment tag attached to all telemetry. | +| `flush_interval` | `float` | `30.0` | Seconds between automatic flushes. | +| `max_batch_size` | `int` | `100` | Early-flush threshold (number of events). | +| `local_port` | `int` | `9847` | WebSocket port for the VS Code extension. | +| `debug` | `bool` | `False` | Log telemetry activity to stderr. | +| `enabled` | `bool` | `True` | Master kill switch — set `False` to disable entirely. | +| `custom_providers` | `list[ProviderDef]` | `[]` | Extra provider rules with higher priority than built-ins. | +| `exclude_patterns` | `list[str]` | `[]` | URL substrings — matching requests are silently dropped. | +| `base_url` | `str` | `"https://api.recost.dev"` | Override for self-hosted deployments. | +| `max_retries` | `int` | `3` | Retry attempts for failed cloud flushes. | +| `on_error` | `Callable` | — | Called on internal SDK errors. | +``` + +Replace with: + +``` +| Option | Type | Default | Description | +|---|---|---|---| +| `api_key` | `str` | — | Recost API key (`rc-...`). If omitted, runs in local mode. | +| `project_id` | `str` | — | Recost project ID. Required in cloud mode. | +| `environment` | `str` | `"development"` | Environment tag attached to all telemetry. | +| `flush_interval_ms` | `int` | `30000` | Milliseconds between automatic aggregator flushes. | +| `flush_interval` | `float` | — | **Deprecated.** Legacy seconds-based flush interval. If set, takes precedence over `flush_interval_ms` and emits a `DeprecationWarning`. Will be removed in a future release. | +| `max_batch_size` | `int` | `100` | Early-flush threshold (number of events). | +| `max_buckets` | `int` | `2000` | Maximum unique (provider, endpoint, method) triplets per window. Crossing this triggers an early flush. | +| `local_port` | `int` | `9847` | WebSocket port for the VS Code extension. | +| `debug` | `bool` | `False` | Log telemetry activity to stderr. | +| `enabled` | `bool` | `True` | Master kill switch — set `False` to disable entirely. | +| `custom_providers` | `list[ProviderDef]` | `[]` | Extra provider rules with higher priority than built-ins. | +| `exclude_patterns` | `list[str]` | `[]` | URL substrings — matching requests are silently dropped. | +| `base_url` | `str` | `"https://api.recost.dev"` | Override for self-hosted deployments. | +| `max_retries` | `int` | `3` | Retry attempts for failed cloud flushes. | +| `shutdown_flush_timeout_ms` | `int` | `3000` | How long `dispose()` waits for the final flush to complete before closing the transport. | +| `on_error` | `Callable[[Exception], None]` | — | Called on internal SDK errors. | +``` + +- [ ] **Step 3: Verify no stale `ReCost` or `flush_interval` (without `_ms`) references remain in code-block examples** + +Run: `python -m grep -n "ReCost\|EcoAPI" README.md` (PowerShell: `Select-String -Path README.md -Pattern "ReCost|EcoAPI"`). + +Expected: only matches inside the deprecation note added in Step 1. + +Run: `python -m grep -n "flush_interval" README.md`. + +Expected: matches are the two new table rows (`flush_interval_ms` and the deprecated `flush_interval`), nothing else. + +- [ ] **Step 4: Commit** + +```bash +git add README.md +git commit -m "docs(README): use RecostExtension; document real config fields + +- Flask example switches to RecostExtension (with a note about the + deprecated ReCost alias). +- Config table swaps the stale 'flush_interval' (seconds, deprecated) + for the canonical 'flush_interval_ms' (milliseconds), with a separate + row marking the old form deprecated. +- Adds previously-undocumented options: max_buckets, + shutdown_flush_timeout_ms. + +Refs #5" +``` + +--- + +## Task 6: Final verification — full suite + open the PR + +- [ ] **Step 1: Run the full test suite** + +Run: `python -m pytest` + +Expected: every test passes (130+ existing, plus the new deprecation tests and the count-pinning assertion). + +- [ ] **Step 2: Verify the deprecation alias actually warns** + +Run: `python -c "from flask import Flask; from recost.frameworks.flask import ReCost; ReCost(Flask(__name__), enabled=False)" 2>&1` + +Expected: output contains `DeprecationWarning: recost.frameworks.flask.ReCost is deprecated ...`. + +(If `flask` is not installed in the dev environment, install it: `python -m pip install -e ".[flask]"`.) + +- [ ] **Step 3: Run ruff and mypy on the touched files** + +Run: `python -m ruff check recost/frameworks/flask.py recost/__init__.py tests/test_flask.py tests/test_provider_registry.py` + +Expected: clean. + +Run: `python -m mypy recost/frameworks/flask.py` + +Expected: no new errors. (Pre-existing mypy errors elsewhere are tracked by issue #2 and out of scope here.) + +- [ ] **Step 4: Push and open the PR** + +```bash +git push -u origin +gh pr create --title "docs+rename: unify naming and reconcile docs (closes #5)" --body "$(cat <<'EOF' +## Summary + +Closes #5. + +- Renames the Flask extension class `ReCost` → `RecostExtension` to match + the convention used by `RecostMiddleware` (FastAPI). `ReCost` remains + importable as a thin subclass that emits a `DeprecationWarning`. +- Strips every reference to nonexistent `EcoAPI*` names from `CLAUDE.md`. +- Replaces the stale `flush_interval` (seconds, deprecated) row in the + README config table with `flush_interval_ms` (canonical) and a + separate deprecated row for `flush_interval`. +- Documents previously-undocumented config fields: `max_buckets`, + `shutdown_flush_timeout_ms`. +- Pins `len(BUILTIN_PROVIDERS) == 34` via a regression test so future + drift trips CI. + +## Tests + +- `TestRecostExtension` — existing Flask tests, updated to the new name. +- `TestReCostDeprecationAlias::test_old_name_still_constructs` — proves + the deprecation alias emits a warning on construction. +- `TestReCostDeprecationAlias::test_old_name_is_subclass_or_alias_of_new` + — proves `isinstance` checks keep working. +- `test_builtin_providers_count_is_pinned` — pins the documented count. + +## Migration + +External consumers using `from recost.frameworks.flask import ReCost` +will see a `DeprecationWarning` but no behavioral change. Switch to +`from recost.frameworks.flask import RecostExtension` before the next +major release. +EOF +)" +``` + +--- + +## Self-review + +- **Spec coverage:** Issue #5 requires (1) pick one brand spelling — done, `Recost`; (2) rename `ReCost` → `RecostExtension` with deprecation alias — Task 1; (3) strip `EcoAPI*` from `CLAUDE.md` — Task 4; (4) update README provider count, flush_interval, missing fields — Task 5; (5) add `len(BUILTIN_PROVIDERS) == 34` test — Task 3. All five accounted for. +- **Placeholder scan:** No TBDs. Every Find/Replace shows exact text. Every test has full body. +- **Type consistency:** `RecostExtension` defined in Task 1, referenced consistently in Task 2 (`__init__.py`), Task 4 (CLAUDE.md description), Task 5 (README example), Task 6 (PR body). `RecostHandle` and `RecostConfig` are imported from `_init`/`_types` and unchanged. +- **Dependencies on other issues:** None. Worktree created from `main`. No conflict with the aggregator plan (different files). From 0322b3745f00ddfe291dbc50628188347e033092 Mon Sep 17 00:00:00 2001 From: Andres Lopez <190146319+AndresL230@users.noreply.github.com> Date: Wed, 13 May 2026 02:58:43 -0400 Subject: [PATCH 5/7] chore(gitignore): ignore .worktrees/ for parallel Wave 1 branches Co-Authored-By: Claude Opus 4.7 (1M context) --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index fce37fb..20b29fd 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,7 @@ __pycache__/ *.egg-info/ dist/ build/ +.worktrees/ # Sensitive .env From a2afd01582196709c7111d87db50de4e5d7056c6 Mon Sep 17 00:00:00 2001 From: Andres Lopez <190146319+AndresL230@users.noreply.github.com> Date: Wed, 13 May 2026 03:06:51 -0400 Subject: [PATCH 6/7] fix(aggregator): make Aggregator thread-safe with RLock Guards _buckets / _window_start / _size with a threading.RLock so the background timer thread's flush() cannot race the interceptor's user threads ingesting events. flush() now uses a swap-and-process pattern so percentile computation runs outside the lock. Adds a stress-test regression that reliably reproduced the race on the old code (RuntimeError: dictionary changed size during iteration). Refs #1 --- recost/_aggregator.py | 85 ++++++++++++++++++++++++---------------- tests/test_aggregator.py | 58 +++++++++++++++++++++++++++ 2 files changed, 109 insertions(+), 34 deletions(-) diff --git a/recost/_aggregator.py b/recost/_aggregator.py index 151f2ac..3e03d39 100644 --- a/recost/_aggregator.py +++ b/recost/_aggregator.py @@ -9,6 +9,7 @@ from __future__ import annotations import math +import threading from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Dict, List, Optional @@ -81,6 +82,10 @@ def __init__( self._buckets: Dict[str, _Bucket] = {} self._window_start: Optional[str] = None self._size = 0 + # RLock so a single thread can re-enter (e.g. if a future event + # callback ever calls back into the aggregator). Guards every method + # below that reads or writes _buckets / _window_start / _size. + self._lock = threading.RLock() # --------------------------------------------------------------------------- # Public API @@ -95,45 +100,60 @@ def _key_for(event: RawEvent) -> str: def would_overflow(self, event: RawEvent) -> bool: """True if ingesting ``event`` would allocate a new bucket while the window is already at capacity. Callers should flush before ingesting.""" - if len(self._buckets) < self._max_buckets: - return False - return self._key_for(event) not in self._buckets + with self._lock: + if len(self._buckets) < self._max_buckets: + return False + return self._key_for(event) not in self._buckets def ingest(self, event: RawEvent, cost_cents: float = 0.0) -> None: """Add one RawEvent to the current window.""" - if self._window_start is None: - self._window_start = event.timestamp + with self._lock: + if self._window_start is None: + self._window_start = event.timestamp - provider = event.provider if event.provider is not None else "unknown" - endpoint = event.endpoint_category if event.endpoint_category is not None else event.path - key = self._key_for(event) + provider = event.provider if event.provider is not None else "unknown" + endpoint = event.endpoint_category if event.endpoint_category is not None else event.path + key = self._key_for(event) - bucket = self._buckets.get(key) - if bucket is None: - bucket = _Bucket(provider=provider, endpoint=endpoint, method=event.method) - self._buckets[key] = bucket + bucket = self._buckets.get(key) + if bucket is None: + bucket = _Bucket(provider=provider, endpoint=endpoint, method=event.method) + self._buckets[key] = bucket - bucket.request_count += 1 - if event.error: - bucket.error_count += 1 - bucket.latencies.append(event.latency_ms) - bucket.total_request_bytes += event.request_bytes - bucket.total_response_bytes += event.response_bytes - bucket.estimated_cost_cents += cost_cents + bucket.request_count += 1 + if event.error: + bucket.error_count += 1 + bucket.latencies.append(event.latency_ms) + bucket.total_request_bytes += event.request_bytes + bucket.total_response_bytes += event.response_bytes + bucket.estimated_cost_cents += cost_cents - self._size += 1 + self._size += 1 def flush(self) -> Optional[WindowSummary]: - """Compress the current window into a WindowSummary and reset state.""" - if not self._buckets: - return None - - window_start = self._window_start or datetime.now(timezone.utc).isoformat() + """Compress the current window into a WindowSummary and reset state. + + Swap-and-process: under the lock, take ownership of the current + buckets dict and reset state; then sort/percentile-compute outside + the lock so ingest is not blocked for the duration of the flush. + """ + with self._lock: + if not self._buckets: + return None + buckets_to_flush = self._buckets + window_start_captured = self._window_start + # Reset state before releasing the lock so concurrent ingests + # land in the *next* window, not this one. + self._buckets = {} + self._window_start = None + self._size = 0 + + # Outside the lock: format timestamps and build the summary. window_end = datetime.now(timezone.utc).isoformat() + window_start = window_start_captured or window_end metrics: List[MetricEntry] = [] - - for bucket in self._buckets.values(): + for bucket in buckets_to_flush.values(): sorted_latencies = sorted(bucket.latencies) total_latency_ms = sum(sorted_latencies) @@ -151,11 +171,6 @@ def flush(self) -> Optional[WindowSummary]: estimated_cost_cents=bucket.estimated_cost_cents, )) - # Reset - self._buckets = {} - self._window_start = None - self._size = 0 - return WindowSummary( project_id=self._project_id, environment=self._environment, @@ -169,12 +184,14 @@ def flush(self) -> Optional[WindowSummary]: @property def size(self) -> int: """Total events ingested since the last flush.""" - return self._size + with self._lock: + return self._size @property def bucket_count(self) -> int: """Number of unique provider + endpoint + method groups.""" - return len(self._buckets) + with self._lock: + return len(self._buckets) @property def max_buckets(self) -> int: diff --git a/tests/test_aggregator.py b/tests/test_aggregator.py index c83d984..b98a3f6 100644 --- a/tests/test_aggregator.py +++ b/tests/test_aggregator.py @@ -4,6 +4,8 @@ Ported from the Node SDK's aggregator.test.ts. """ +import sys +import threading from datetime import datetime, timezone from recost._aggregator import Aggregator, MAX_BUCKETS @@ -380,3 +382,59 @@ def test_default_cap_fires_at_2001st_triplet(self): assert agg.bucket_count == 2000 overflow = make_event(provider="p2000", endpoint_category="ep2000") assert agg.would_overflow(overflow) + + +# --------------------------------------------------------------------------- +# Thread safety +# --------------------------------------------------------------------------- + + +class TestThreadSafety: + """Regression tests for issue #1 — Aggregator must be safe under concurrent + ingest (user threads) + flush (timer thread).""" + + def test_concurrent_ingest_and_flush_does_not_raise(self): + """Stress: 4 ingester threads run 50000 ingests each (200k total) while + one flusher thread runs 500 flushes. Uses ``sys.setswitchinterval`` to + force aggressive GIL handoff so the race window is reliably hit. + Pre-fix this raises ``RuntimeError: dictionary changed size during + iteration``. Post-fix it must complete without any exception.""" + agg = Aggregator() + exceptions: list[BaseException] = [] + iterations_per_ingester = 50000 + num_ingesters = 4 + num_flushes = 500 + + def ingester() -> None: + try: + for i in range(iterations_per_ingester): + p = f"p{i % 100}" + agg.ingest(make_event(provider=p, endpoint_category=p)) + except BaseException as exc: # noqa: BLE001 — we want everything + exceptions.append(exc) + + def flusher() -> None: + try: + for _ in range(num_flushes): + agg.flush() + except BaseException as exc: # noqa: BLE001 + exceptions.append(exc) + + # Force aggressive thread switching so the race window is hit reliably + # on modern CPython where short Python operations rarely yield the GIL. + original_interval = sys.getswitchinterval() + sys.setswitchinterval(0.000001) + try: + threads = [threading.Thread(target=ingester) for _ in range(num_ingesters)] + threads.append(threading.Thread(target=flusher)) + for t in threads: + t.start() + for t in threads: + t.join(timeout=30.0) + finally: + sys.setswitchinterval(original_interval) + + assert not exceptions, f"unexpected exceptions: {exceptions!r}" + # Sanity: at least some threads ran to completion. + for t in threads: + assert not t.is_alive(), "a worker thread is still running" From f055ae23267e6513cc7d5ec2932ba79e32bd646f Mon Sep 17 00:00:00 2001 From: Andres Lopez <190146319+AndresL230@users.noreply.github.com> Date: Wed, 13 May 2026 03:10:15 -0400 Subject: [PATCH 7/7] test(aggregator): assert concurrent ingest does not lose events Adds a second thread-safety test that proves no counter increments are lost when N threads race on a shared bucket. Complements the no-raise stress test by exercising the counter-increment side of the race. Refs #1 --- tests/test_aggregator.py | 64 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 64 insertions(+) diff --git a/tests/test_aggregator.py b/tests/test_aggregator.py index b98a3f6..3603f72 100644 --- a/tests/test_aggregator.py +++ b/tests/test_aggregator.py @@ -438,3 +438,67 @@ def flusher() -> None: # Sanity: at least some threads ran to completion. for t in threads: assert not t.is_alive(), "a worker thread is still running" + + def test_concurrent_correctness_no_lost_events(self): + """4 ingester threads each push 2000 events while a flusher pulls + windows concurrently. The total request_count summed across every + flushed summary, plus the request_count of the final drain, must + equal the total number of events ingested. Pre-fix this fails + because ``bucket.request_count += 1`` races across threads.""" + agg = Aggregator() + iterations_per_ingester = 2000 + num_ingesters = 4 + total_expected = iterations_per_ingester * num_ingesters + flushed_total = 0 + flushed_lock = threading.Lock() # this lock is in the *test*, not the SUT + exceptions: list[BaseException] = [] + ingest_done = threading.Event() + + def ingester() -> None: + try: + for i in range(iterations_per_ingester): + # Small key space so several events land in the same + # bucket and the counter increment is contested. + p = f"p{i % 3}" + agg.ingest(make_event(provider=p, endpoint_category=p)) + except BaseException as exc: # noqa: BLE001 + exceptions.append(exc) + + def flusher() -> None: + nonlocal flushed_total + try: + while not ingest_done.is_set(): + summary = agg.flush() + if summary is not None: + n = sum(m.request_count for m in summary.metrics) + with flushed_lock: + flushed_total += n + except BaseException as exc: # noqa: BLE001 + exceptions.append(exc) + + # Force aggressive GIL yields so the counter race fires reliably + # on modern CPython. Restored in finally. + original_switch = sys.getswitchinterval() + sys.setswitchinterval(0.000001) + try: + ingesters = [threading.Thread(target=ingester) for _ in range(num_ingesters)] + flusher_thread = threading.Thread(target=flusher) + flusher_thread.start() + for t in ingesters: + t.start() + for t in ingesters: + t.join(timeout=30.0) + ingest_done.set() + flusher_thread.join(timeout=10.0) + finally: + sys.setswitchinterval(original_switch) + + # Final drain — anything ingested after the last flusher iteration. + final = agg.flush() + if final is not None: + flushed_total += sum(m.request_count for m in final.metrics) + + assert not exceptions, f"unexpected exceptions: {exceptions!r}" + assert flushed_total == total_expected, ( + f"lost events: flushed {flushed_total}, expected {total_expected}" + )