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 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**. 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). 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" } 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..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 @@ -16,6 +17,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,53 +73,87 @@ 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 + # 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 # --------------------------------------------------------------------------- + @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.""" + 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 = f"{provider}::{endpoint}::{event.method}" + 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) @@ -124,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, @@ -142,9 +184,16 @@ 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: + """Configured maximum buckets per window.""" + return self._max_buckets diff --git a/tests/test_aggregator.py b/tests/test_aggregator.py index e682dee..3603f72 100644 --- a/tests/test_aggregator.py +++ b/tests/test_aggregator.py @@ -4,9 +4,11 @@ Ported from the Node SDK's aggregator.test.ts. """ +import sys +import threading from datetime import datetime, timezone -from recost._aggregator import Aggregator +from recost._aggregator import Aggregator, MAX_BUCKETS from recost._types import RawEvent @@ -337,3 +339,166 @@ 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) + + +# --------------------------------------------------------------------------- +# 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" + + 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}" + ) 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()