feat(proxy): Redis hot-path metering with SQLite rehydration

[oblangatas]

Summary

• Adds Redis as an optional hot-path enforcer in front of SQLite spend_log. SpendCapRule does GET worthless:spend:{alias} before XOR reconstruction; a miss rehydrates from SELECT SUM(tokens) FROM spend_log + SET NX. SQLite remains the authoritative ledger.
• Redis is optional: unset WORTHLESS_REDIS_URL → pre-Redis behaviour (no regression).
• Compose gains a redis:7-alpine service on an internal-only backend network with noeviction, no persistence, tight resource caps.

Why this shape (gate-before-reconstruct, SR-03)

The original implementation was reviewed by four agents in parallel (security, architect, brutus, chaos) before any commit landed. Six block-ship findings were addressed in the first commit:

Finding	Fix
Cache miss / cold start / restart / eviction → counter reads 0 → silent cap bypass	Miss rehydrates from SQLite SUM + SET NX
allkeys-lru evicts hot counters under pressure → silent reset	noeviction
No socket timeout → slow Redis parks the gate handler indefinitely	socket_timeout=2s, socket_connect_timeout=1s, health_check_interval=30s
Malformed Redis value → int() returns 0 → fail-open	RedisValueError → treat as miss, rehydrate
Redis transport error → 402-storm on every capped alias	Fall back to _evaluate_sqlite (SR-03 still holds — reconstruction gates on return value)
Sentinel-rule invariant test only proved engine short-circuit	Rewritten as TestClient through ASGI, patches reconstruct_key / reconstruct_key_fp on worthless.proxy.app, asserts await_count == 0

Startup hardening: create_redis_client validates URL scheme (redis:// / rediss:// only — compromised env cannot redirect to unix:// or file://), pings at boot so typo'd URLs fail at startup rather than on first request.

Tests (48 new)

• Unit (test_redis_metering.py, 27): stub-based coverage of every code path, plus two TestClient invariant tests (denial + Redis-outage fallback both assert reconstruct never called).
• Dynamic (test_redis_metering_dynamic.py, 21):
  • 16 fakeredis tests drive the real redis.asyncio.Redis client against an in-process FakeRedis server — catches protocol bugs a stub hides (SET NX semantics, INCRBY atomicity under asyncio.gather).
  • 5 docker-gated tests spin redis:7-alpine (session-scoped) and exercise SpendCapRule, record_spend, create_redis_client end-to-end over real TCP. Skipped when docker is unavailable and WORTHLESS_TEST_REDIS_URL is unset.

All 314 tests in the relevant suites pass. No regressions in proxy/e2e/hardening.

Deferred to follow-up PRs

• Redis AUTH + TLS (rediss://, requirepass, rename-command FLUSHALL "")
• Circuit breaker for Redis flaps
• Reconciler for partial-write counter drift (SQLite insert succeeded, Redis INCR failed)
• Benchmark justifying the dependency at v1.1's target RPS (may defer the full feature to v2.0 if SQLite is fast enough)

Test plan

• uv run pytest tests/test_redis_metering.py tests/test_redis_metering_dynamic.py
• Spin the compose stack and hit the proxy with a capped alias: verify noeviction policy via docker compose exec redis redis-cli config get maxmemory-policy
• Unset WORTHLESS_REDIS_URL and confirm the full existing test matrix still passes (no-regression)
• Simulate a Redis outage (docker compose stop redis) with an under-cap alias — request should still succeed via SQLite fallback

🤖 Generated with Claude Code

Summary by CodeRabbit

• New Features

  • Optional Redis hot-path metering for faster spend-cap checks (SQLite remains authoritative); opt-in via WORTHLESS_REDIS_URL and runtime toggle.

• Bug Fixes

  • Prevents reservation leaks by removing fully-released entries instead of leaving zeroed records.

• Documentation

  • Added benchmark report (SQLite vs Redis), deployment guidance, README and env example updates with opt-in criteria.

• Tests

  • Comprehensive unit, dynamic/docker-gated, failure-mode, property, integration, streaming, and benchmark suites for Redis metering.

• Chores

  • Docker Compose adds an optional hardened Redis service; packaging exposes Redis optional dependency.

[coderabbitai]

Warning

Rate limit exceeded

@shacharm2 has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 18 minutes and 13 seconds before requesting another review.

Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 18 minutes and 13 seconds.

⌛ How to resolve this issue?

After the wait time has elapsed, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout.

Please see our FAQ for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: be33add9-e83e-4a09-8450-feabcd9a0ac2

📥 Commits

Reviewing files that changed from the base of the PR and between 7dfb79e and f0001d3.

📒 Files selected for processing (13)

• .planning/bench/spend-cap-rule-sqlite-vs-redis.md
• .planning/c3ox-design.md
• README.md
• deploy/docker-compose.env.example
• deploy/docker-compose.yml
• deploy/redis-acl.conf.tmpl
• pyproject.toml
• src/worthless/proxy/rules.py
• tests/test_deploy_static.py
• tests/test_redis_metering.py
• tests/test_redis_metering_defense_in_depth.py
• tests/test_streaming_redis_failure_modes.py
• tests/test_streaming_redis_integration.py

📝 Walkthrough

Walkthrough

Adds an opt-in Redis hot-path layered over SQLite for spend-cap metering, wiring lifecycle/config, dual-phase recording (SQLite authoritative + best-effort Redis INCR), Redis-aware SpendCapRule with rehydrate/dirty-tracker logic, deployment/docker changes, benchmarks, and many new tests covering correctness and failure modes.

Changes

Cohort / File(s)	Summary
Documentation & Benchmarks .planning/bench/spend-cap-rule-sqlite-vs-redis.md, README.md	Adds benchmark report and README opt-in guidance describing Redis hot-path semantics, measured latency observations, enablement criteria, and re-run instructions.
Docker / Deployment Dockerfile, deploy/docker-compose.yml, deploy/docker-compose.env.example	Adds Redis service to compose (no persistence, 128MB cap, noeviction), hardening/healthcheck; documents WORTHLESS_REDIS_URL opt-in; Dockerfile installs .[redis].
Dependency & Tooling pyproject.toml, .pre-commit-config.yaml	Adds redis optional deps and fakeredis to test deps, pytest redis marker and Ruff ignores for docker-invoking tests; suppresses specific pip-audit vuln in pre-commit hook.
App Lifecycle & Config src/worthless/proxy/app.py, src/worthless/proxy/config.py	Wires optional Redis client and SpendDirtyTracker into FastAPI lifespan (app.state), exposes redis_url setting, threads redis/dirty-tracker into record paths, and best-effort closes client on shutdown.
Metering Core & Rules src/worthless/proxy/metering.py, src/worthless/proxy/rules.py	Implements Redis client creation/validation, hot-counter helpers (spend_key, get_spend_hot, incr_spend_hot, rehydrate_spend_hot), RedisValueError, SpendDirtyTracker, dual-phase record_spend (SQLite authoritative + best-effort Redis INCRBY), Redis-aware SpendCapRule with rehydrate-on-miss and SQLite fallback, and trims zero entries from reservation maps on release.
Benchmarks & Tests tests/bench_spend_cap_rule.py, tests/test_redis_metering*.py, tests/test_streaming_redis_*.py, tests/conftest.py	Adds benchmark harness and extensive unit/dynamic/property/integration/failure-mode tests (many docker-gated) exercising hot-counter semantics, rehydration, malformed-value handling, timeouts/failures, concurrency, reservation invariants, and auto-marking of Redis tests.

Sequence Diagram(s)

sequenceDiagram
    participant Client
    participant Proxy as Proxy<br/>(SpendCapRule)
    participant Redis
    participant SQLite

    Client->>Proxy: request (spend-cap check)
    alt Redis configured
        Proxy->>Redis: GET spend:alias
        alt Cache hit (valid int)
            Redis-->>Proxy: counter
            Proxy->>Proxy: check counter + reserved vs cap
        else Cache miss / malformed
            Proxy->>SQLite: SELECT SUM(tokens) FROM spend_log
            SQLite-->>Proxy: ledger sum
            Proxy->>Redis: SET NX spend:alias ledger_sum
            Proxy->>Proxy: check ledger_sum + reserved vs cap
        else Redis transport error
            Proxy->>SQLite: BEGIN IMMEDIATE (fallback)
            SQLite-->>Proxy: lock acquired
            Proxy->>Proxy: check SQLite state + reserved vs cap
        end
    else No Redis
        Proxy->>SQLite: BEGIN IMMEDIATE
        SQLite-->>Proxy: lock acquired
        Proxy->>Proxy: check SQLite state + reserved vs cap
    end

    alt Allowed
        Proxy-->>Client: 200 OK
        Proxy->>SQLite: INSERT INTO spend_log (durable)
        opt tokens > 0
            Proxy->>Redis: INCRBY spend:alias tokens (best-effort)
            Redis-->>Proxy: reply / error
        end
    else Denied
        Proxy-->>Client: 402 Payment Required
    end

Loading

Estimated Code Review Effort

🎯 4 (Complex) | ⏱️ ~50 minutes

Possibly related PRs

• fix: V1 release blockers — 7 P1 security + correctness fixes #46 — Overlaps proxy lifecycle and metering changes touching src/worthless/proxy/app.py, metering.py, and rules behavior.

Poem

🐰 I hopped through code with counters bright,
I cached quick hops while ledger stayed right,
SQLite keeps truth in tidy rows,
Redis hums when concurrent wind blows,
A carrot for fast-path delight 🥕

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 39.51% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly and specifically summarizes the main feature: adding Redis hot-path metering while maintaining SQLite as the authoritative ledger with rehydration capability.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/redis-metering

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 4

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

deploy/docker-compose.yml (1)

2-47: ⚠️ Potential issue | 🟡 Minor

Add depends_on so the proxy doesn't crashloop on cold starts when Redis is enabled.

With WORTHLESS_REDIS_URL set, create_redis_client issues PING during _lifespan and propagates failures out of FastAPI startup. Because there is no depends_on linking the proxy to the redis service, Docker Compose will start both in parallel — on a cold docker compose up, the proxy can finish startup before redis is accepting connections, fail the ping, and crashloop via restart: unless-stopped until redis is healthy.

The redis service already exposes a healthcheck (lines 86-90), so this is a one-line fix.

🔧 Proposed fix

     image: redis:7-alpine
     # ...
   proxy:
     build:
       context: ..
       dockerfile: Dockerfile
     ports:
       - "127.0.0.1:8787:8787"
     env_file: docker-compose.env
+    depends_on:
+      redis:
+        condition: service_healthy

Alternatively, if you want the proxy to stay startable without redis (Redis disabled path), gate this behind a compose profile so depends_on is only active when the redis profile is selected.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@deploy/docker-compose.yml` around lines 2 - 47, The proxy can start before
Redis is ready causing create_redis_client (used during _lifespan when
WORTHLESS_REDIS_URL is set) to PING and fail; add a Docker Compose service
dependency so the proxy waits for Redis' healthcheck. In the proxy service block
(named "proxy") add a depends_on entry referencing the "redis" service and
require the health check (service_healthy) so Compose will wait for redis'
healthcheck to pass before starting the proxy; alternatively gate this
depends_on behind a compose profile if you need the proxy to start without
Redis.

🧹 Nitpick comments (5)

src/worthless/proxy/app.py (1)

207-215: Optional: isolate each close step so one failure doesn't skip the remaining resources.

client.aclose() currently runs outside a try/except in the outer block. If it raises, redis_client.aclose(), db.close(), and repo.close() are all skipped (the finally still zeroes the Fernet key, so SR-02 holds). The new redis close already uses this isolated pattern — consider extending it to the others for symmetric cleanup.

♻️ Example — isolate each shutdown step

 try:
-    await client.aclose()
-    if redis_client is not None:
-        try:
-            await redis_client.aclose()
-        except Exception:  # noqa: S110 — best-effort close; do not mask other shutdown errors  # nosec B110
-            pass
-    await db.close()
-    repo.close()
+    for step in (
+        client.aclose,
+        (redis_client.aclose if redis_client is not None else None),
+        db.close,
+        repo.close,
+    ):
+        if step is None:
+            continue
+        try:
+            result = step()
+            if asyncio.iscoroutine(result):
+                await result
+        except Exception:  # noqa: S110 — best-effort close  # nosec B110
+            logger.warning("shutdown step %s raised", getattr(step, "__qualname__", step))
 finally:
     for i in range(len(settings.fernet_key)):
         settings.fernet_key[i] = 0

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/worthless/proxy/app.py` around lines 207 - 215, The shutdown sequence
currently calls client.aclose() outside an isolated try/except so a failure
there prevents redis_client.aclose(), db.close(), and repo.close() from running;
wrap each close call (client.aclose, redis_client.aclose already isolated,
db.close, repo.close) in its own try/except so each resource gets a best-effort
close regardless of others, log or swallow exceptions consistently (following
the existing pattern and comments about best-effort closes) and preserve the
surrounding finally block that zeroes the Fernet key.

deploy/docker-compose.yml (1)

50-50: Optional: pin redis:7-alpine by digest to match Dockerfile supply-chain policy.

The Dockerfile pins python:3.13-slim-bookworm@sha256:..., but the redis image uses a floating tag. A digest pin (redis:7-alpine@sha256:...) aligns the two and removes a silent-upgrade vector during docker compose pull.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@deploy/docker-compose.yml` at line 50, Replace the floating tag "image:
redis:7-alpine" with a digest-pinned image (e.g.,
"redis:7-alpine@sha256:<digest>") to match the Dockerfile supply-chain policy;
obtain the correct sha256 digest for redis:7-alpine from the official registry
(docker pull + docker inspect or Docker Hub / registry manifest) and update the
image reference so pulls use the exact immutable image digest.

tests/bench_spend_cap_rule.py (1)

118-128: Dead code: _run_async is never called.

Every benchmark inlines asyncio.new_event_loop() + loop.run_until_complete(...) directly (e.g., lines 141-147, 159-165). The helper's finally block is also a no-op (pass # loop closed by the caller). Remove it, or consolidate the duplicated loop-management boilerplate through it.

♻️ Option A — remove the helper

-def _run_async(coro_factory):
-    """pytest-benchmark adapter — wrap an async callable for sync benchmarking.
-
-    We explicitly reuse one event loop per benchmark so fixture/loop
-    overhead doesn't bias the timing.
-    """
-    loop = asyncio.new_event_loop()
-    try:
-        return lambda: loop.run_until_complete(coro_factory())
-    finally:
-        pass  # loop closed by the caller after all iterations
-
-

♻️ Option B — actually use it (removes ~6 copies of boilerplate)

 def test_bench_single_sqlite(benchmark, seeded):
     rule_sqlite, _rule_redis, alias, rows = seeded
 
     async def one():
         body = b'{"model":"gpt-4","max_tokens":100}'
         await rule_sqlite.evaluate(alias, object(), provider="openai", body=body)
         await rule_sqlite.release_reservation(alias, 100)
 
-    loop = asyncio.new_event_loop()
-    try:
-        benchmark.extra_info["rows"] = rows
-        benchmark.extra_info["backend"] = "sqlite"
-        benchmark(lambda: loop.run_until_complete(one()))
-    finally:
-        loop.close()
+    benchmark.extra_info["rows"] = rows
+    benchmark.extra_info["backend"] = "sqlite"
+    with _benchmark_loop() as run:
+        benchmark(lambda: run(one()))

…with a small context-manager helper that actually closes the loop.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/bench_spend_cap_rule.py` around lines 118 - 128, The helper function
_run_async is dead and currently a no-op in its finally block while multiple
benchmarks duplicate asyncio.new_event_loop()/loop.run_until_complete(...)
boilerplate; fix this by implementing Option B: update _run_async to create an
event loop, return a callable that runs coro_factory() on that loop and closes
the loop after all iterations (or provide a small context-manager variant that
ensures loop.close()), then replace the repeated inline loop
creation/run_until_complete blocks in the benchmarks with calls to
_run_async(coro_factory) so all loop management is centralized (refer to the
symbol _run_async and the existing inline uses of
asyncio.new_event_loop()/loop.run_until_complete in the benchmark functions).

tests/test_redis_metering.py (1)

420-423: Two readability nits: mid-file import asyncio and a dead-conditional alias.

• Line 422: import asyncio # noqa: E402 is deferred ~400 lines down and silences E402. Move to the top import block alongside the other stdlib imports; no need for noqa.
• Line 565: rule._reserved["alias" if False else "alice"] evaluates unconditionally to "alice". Looks like a leftover from refactoring. Just use the literal key.

✏️ Proposed fixes

-from typing import Any
+import asyncio
+from typing import Any
 from unittest.mock import AsyncMock

-# ---------------------------------------------------------------------------
-# WOR-242 × Redis — reservation mechanism across the hot path.
-# ...
-# ---------------------------------------------------------------------------
-
-
-import asyncio  # noqa: E402
-
-
 `@pytest.mark.asyncio`

-    # Invariant: cumulative reservation equals cap (800 + 200 = 1000).
-    assert rule._reserved["alias" if False else "alice"] == 1000
+    # Invariant: cumulative reservation equals cap (800 + 200 = 1000).
+    assert rule._reserved["alice"] == 1000

Also applies to: 562-570

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_redis_metering.py` around lines 420 - 423, Move the mid-file
"import asyncio  # noqa: E402" into the top-level stdlib import block with the
other imports (remove the noqa) to fix E402 and improve readability, and replace
the dead-conditional access rule._reserved["alias" if False else "alice"] with
the literal key rule._reserved["alice"] (and any other occurrences in the
562-570 range) to remove the leftover refactor artifact.

src/worthless/proxy/rules.py (1)

135-188: Redis gate path looks correct and preserves SR-03 semantics.

Three-way branching (int / RedisValueError-as-miss / transport error → SQLite fallback) is clearly implemented, and the reservation block correctly runs under _reserve_lock using the committed counter from Redis or rehydrate. A few things worth noting for defense-in-depth, but none are blocking:

• _evaluate_redis lacks the outer try: … except Exception: return spend_cap_error_response(...) catch-all that _evaluate_sqlite has. Everything inside is already guarded, but an unexpected error in the reservation math (e.g. a malformed spend_cap column value) would propagate as a 500 rather than a fail-closed 402. Worth considering a belt-and-braces wrap.
• spend_cap is compared as a float and only cast via int(spend_cap) when computing remaining. Fine today, but if caps ever allow fractional values this truncates remaining downward. Matches pre-existing SQLite-path behaviour.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/worthless/proxy/rules.py` around lines 135 - 188, _wrap the entire body
of _evaluate_redis in a top-level try/except that catches Exception and returns
spend_cap_error_response(provider=provider) to mirror _evaluate_sqlite's
fail-closed behavior; keep the existing inner catches (RedisValueError,
transport fallback, rehydrate handling) but ensure any unexpected errors (e.g.,
malformed spend_cap, arithmetic issues during reservation under _reserve_lock,
or issues with _reserved) cause a 402 via spend_cap_error_response rather than
propagating a 500. Use the same provider argument when calling
spend_cap_error_response to match existing semantics.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In @.planning/bench/spend-cap-rule-sqlite-vs-redis.md:
- Around line 103-108: The fenced code block that contains the pytest re-run
command is missing a language identifier (MD040); update the opening
triple-backtick to include a language such as bash (e.g., replace ``` with
```bash) so the block reads ```bash and preserves the same command lines (pytest
tests/bench_spend_cap_rule.py ... --benchmark-json=/tmp/bench.json).

In `@src/worthless/proxy/metering.py`:
- Around line 78-88: If Redis.from_url(...) in create_redis_client succeeds but
await client.ping() fails, the client and its connection pool must be closed to
avoid leaking resources; wrap the ping call in a try/except, and on any
exception call the client's close/disconnect coroutine (e.g., await
client.close() or await client.connection_pool.disconnect() as appropriate for
the redis async client) before re-raising the exception so the pool is cleaned
up; keep references to Redis.from_url, create_redis_client, and client.ping to
locate the change.

In `@tests/bench_spend_cap_rule.py`:
- Around line 320-358: The seeded_tcp fixture binds a redis-py async client to
pytest-asyncio's loop causing "Future attached to a different loop" when
benchmarks create their own event loop; change seeded_tcp to be synchronous/lazy
so the actual create_redis_client call happens inside the benchmark's loop (i.e.
convert async def seeded_tcp -> def seeded_tcp and yield a factory like
_make_redis that the test calls inside its event loop to await
create_redis_client(tcp_redis_url), then call incr_spend_hot(redis, "alice",
...) and construct SpendCapRule(db=redis_db, redis=redis) there); ensure the
factory returns the same tuple (rule, "alice", size) or returns the redis and
spend_key/alice so the test can create SpendCapRule and perform proper async
cleanup (close redis with redis.aclose() and sqlite with redis_db.close()) from
the benchmark loop.

In `@tests/test_redis_metering_dynamic.py`:
- Around line 542-555: Modify test_real_redis_noeviction_policy_is_set so it
does not fail when pointing at an external Redis via WORTHLESS_TEST_REDIS_URL:
after fetching policy from real_redis_client (in
test_real_redis_noeviction_policy_is_set), check
os.environ.get("WORTHLESS_TEST_REDIS_URL") (or a flag on the real_redis_client
fixture if available) and if the env var is set and the decoded value is not
"noeviction", call pytest.skip with a clear message; ensure pytest and os are
imported and leave the existing docker_required decorator and real_redis_client
usage unchanged.

---

Outside diff comments:
In `@deploy/docker-compose.yml`:
- Around line 2-47: The proxy can start before Redis is ready causing
create_redis_client (used during _lifespan when WORTHLESS_REDIS_URL is set) to
PING and fail; add a Docker Compose service dependency so the proxy waits for
Redis' healthcheck. In the proxy service block (named "proxy") add a depends_on
entry referencing the "redis" service and require the health check
(service_healthy) so Compose will wait for redis' healthcheck to pass before
starting the proxy; alternatively gate this depends_on behind a compose profile
if you need the proxy to start without Redis.

---

Nitpick comments:
In `@deploy/docker-compose.yml`:
- Line 50: Replace the floating tag "image: redis:7-alpine" with a digest-pinned
image (e.g., "redis:7-alpine@sha256:<digest>") to match the Dockerfile
supply-chain policy; obtain the correct sha256 digest for redis:7-alpine from
the official registry (docker pull + docker inspect or Docker Hub / registry
manifest) and update the image reference so pulls use the exact immutable image
digest.

In `@src/worthless/proxy/app.py`:
- Around line 207-215: The shutdown sequence currently calls client.aclose()
outside an isolated try/except so a failure there prevents
redis_client.aclose(), db.close(), and repo.close() from running; wrap each
close call (client.aclose, redis_client.aclose already isolated, db.close,
repo.close) in its own try/except so each resource gets a best-effort close
regardless of others, log or swallow exceptions consistently (following the
existing pattern and comments about best-effort closes) and preserve the
surrounding finally block that zeroes the Fernet key.

In `@src/worthless/proxy/rules.py`:
- Around line 135-188: _wrap the entire body of _evaluate_redis in a top-level
try/except that catches Exception and returns
spend_cap_error_response(provider=provider) to mirror _evaluate_sqlite's
fail-closed behavior; keep the existing inner catches (RedisValueError,
transport fallback, rehydrate handling) but ensure any unexpected errors (e.g.,
malformed spend_cap, arithmetic issues during reservation under _reserve_lock,
or issues with _reserved) cause a 402 via spend_cap_error_response rather than
propagating a 500. Use the same provider argument when calling
spend_cap_error_response to match existing semantics.

In `@tests/bench_spend_cap_rule.py`:
- Around line 118-128: The helper function _run_async is dead and currently a
no-op in its finally block while multiple benchmarks duplicate
asyncio.new_event_loop()/loop.run_until_complete(...) boilerplate; fix this by
implementing Option B: update _run_async to create an event loop, return a
callable that runs coro_factory() on that loop and closes the loop after all
iterations (or provide a small context-manager variant that ensures
loop.close()), then replace the repeated inline loop creation/run_until_complete
blocks in the benchmarks with calls to _run_async(coro_factory) so all loop
management is centralized (refer to the symbol _run_async and the existing
inline uses of asyncio.new_event_loop()/loop.run_until_complete in the benchmark
functions).

In `@tests/test_redis_metering.py`:
- Around line 420-423: Move the mid-file "import asyncio  # noqa: E402" into the
top-level stdlib import block with the other imports (remove the noqa) to fix
E402 and improve readability, and replace the dead-conditional access
rule._reserved["alias" if False else "alice"] with the literal key
rule._reserved["alice"] (and any other occurrences in the 562-570 range) to
remove the leftover refactor artifact.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 1685ea82-cf59-4245-b191-808a8eb6007e

📥 Commits

Reviewing files that changed from the base of the PR and between 0c69d68 and cf1fc2e.

⛔ Files ignored due to path filters (1)

• uv.lock is excluded by !**/*.lock

📒 Files selected for processing (13)

• .planning/bench/spend-cap-rule-sqlite-vs-redis.md
• Dockerfile
• README.md
• deploy/docker-compose.env.example
• deploy/docker-compose.yml
• pyproject.toml
• src/worthless/proxy/app.py
• src/worthless/proxy/config.py
• src/worthless/proxy/metering.py
• src/worthless/proxy/rules.py
• tests/bench_spend_cap_rule.py
• tests/test_redis_metering.py
• tests/test_redis_metering_dynamic.py

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

🧩 Analysis chain

🌐 Web query:

redis-py redis.asyncio Redis.from_url lazy connection pool close on failure

💡 Result:

In redis-py redis.asyncio, Redis.from_url("redis://...") creates a lazy connection pool (ConnectionPool or BlockingConnectionPool) that does not establish connections upfront. Connections are created on-demand when acquired via get_connection and released back after use. On failure (e.g., ConnectionError, TimeoutError during connect, send_command, or read_response), the connection is explicitly disconnected via connection.disconnect [3,23,26]. Faulty connections are not returned to the pool's available queue; instead: - In BlockingConnectionPool.get_connection, if await self.ensure_connection(connection) fails with BaseException, it calls await self.release(connection) before re-raising [3]. - The client execute_command wraps operations with retry.call_with_retry, which on retry calls _disconnect_raise_connect, invoking connection.disconnect [18,25]. - Connection.send_packed_command and read_response catch errors (OSError, BaseException) and call disconnect before raising [3,23]. This prevents stale/bad connections from being reused. The pool creates new connections as needed up to max_connections. Health checks (health_check_interval > 0) proactively PING idle connections, disconnecting on failure [14]. Use BlockingConnectionPool for production to avoid ConnectionError when pool is exhausted [3]. Always await client.aclose to disconnect the pool [5,9].

Citations:

• 1: https://github.com/redis/redis-py/blob/master/redis/asyncio/connection.py
• 2: https://redis.readthedocs.io/en/v6.2.0/examples/asyncio_examples.html
• 3: https://redis.readthedocs.io/en/stable/examples/asyncio_examples.html
• 4: https://redis.io/docs/latest/develop/clients/redis-py/produsage
• 5: [Bug] Clients running with connection pool configured cannot retry on transient network failures redis/redis-py#1789
• 6: https://github.com/redis/redis-py/blob/master/redis/asyncio/client.py
• 7: Connection class disconnects on BaseException redis/redis-py#2103
• 8: Inconsistency default behaviour on Sync / Async clients against ConnectionError redis/redis-py#3194
• 9: https://redis.readthedocs.io/en/v6.1.0/_modules/redis/connection.html
• 10: BlockingConnectionPool does not recover if redis disconnects. redis/redis-py#3034
• 11: BlockingConnectionPool deadlock (double condition.acquire) redis/redis-py#3056

---

🏁 Script executed:

# First, let's look at the actual file to understand the function context
cat -n src/worthless/proxy/metering.py | head -100

Repository: shacharm2/worthless

Length of output: 4577

---

🏁 Script executed:

# Search for how Redis clients are used elsewhere in the codebase
rg -t py "aclose\(\)" --context 3

Repository: shacharm2/worthless

Length of output: 11816

---

🏁 Script executed:

# Check if there are other Redis client initialization patterns
rg -t py "Redis\.from_url\|client\.aclose\|client\.close" --context 2

Repository: shacharm2/worthless

Length of output: 45

---

🏁 Script executed:

# Search for tests of create_redis_client, especially error paths
rg -t py "create_redis_client" --context 5

Repository: shacharm2/worthless

Length of output: 14080

---

🏁 Script executed:

# Check how create_redis_client is called in the codebase
rg -t py "create_redis_client\(" --context 3

Repository: shacharm2/worthless

Length of output: 4534

---

Close the client on ping failure in create_redis_client to avoid leaking the connection pool.

If Redis.from_url(...) succeeds but await client.ping() raises (unreachable host, auth failure, protocol mismatch), the constructed client and its connection pool are never closed before the exception unwinds. On startup attempts against a misconfigured URL, this leaks the pool's resources until garbage collection eventually reaps them.

🔧 Proposed fix

     client = Redis.from_url(
         url,
         decode_responses=False,
         socket_timeout=_REDIS_SOCKET_TIMEOUT,
         socket_connect_timeout=_REDIS_CONNECT_TIMEOUT,
         health_check_interval=30,
     )
-    # redis-py's async type stubs mark ping() as returning bool, not
-    # Awaitable[bool] — it *is* a coroutine at runtime. Cast away.
-    await client.ping()  # type: ignore[misc]
-    return client
+    try:
+        # redis-py's async type stubs mark ping() as returning bool, not
+        # Awaitable[bool] — it *is* a coroutine at runtime. Cast away.
+        await client.ping()  # type: ignore[misc]
+    except Exception:
+        await client.aclose()
+        raise
+    return client

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	client = Redis.from_url(
	url,
	decode_responses=False,
	socket_timeout=_REDIS_SOCKET_TIMEOUT,
	socket_connect_timeout=_REDIS_CONNECT_TIMEOUT,
	health_check_interval=30,
	)
	# redis-py's async type stubs mark ping() as returning bool, not
	# Awaitable[bool] — it *is* a coroutine at runtime. Cast away.
	await client.ping() # type: ignore[misc]
	return client
	client = Redis.from_url(
	url,
	decode_responses=False,
	socket_timeout=_REDIS_SOCKET_TIMEOUT,
	socket_connect_timeout=_REDIS_CONNECT_TIMEOUT,
	health_check_interval=30,
	)
	try:
	# redis-py's async type stubs mark ping() as returning bool, not
	# Awaitable[bool] — it *is* a coroutine at runtime. Cast away.
	await client.ping() # type: ignore[misc]
	except Exception:
	await client.aclose()
	raise
	return client

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/worthless/proxy/metering.py` around lines 78 - 88, If Redis.from_url(...)
in create_redis_client succeeds but await client.ping() fails, the client and
its connection pool must be closed to avoid leaking resources; wrap the ping
call in a try/except, and on any exception call the client's close/disconnect
coroutine (e.g., await client.close() or await
client.connection_pool.disconnect() as appropriate for the redis async client)
before re-raising the exception so the pool is cleaned up; keep references to
Redis.from_url, create_redis_client, and client.ping to locate the change.

[oblangatas]

Acknowledged but no change. Per the linked redis-py research: Redis.from_url is lazy (no upfront connect), and on PING failure inside create_redis_client the exception propagates without leaving an open pool — the lazy pool only allocates connections on first command, and PING is the first command. Worst case is one half-open socket if PING raises mid-handshake, which redis-py's disconnect_on_error retry logic releases. I considered wrapping the PING in try/except + explicit await client.aclose() on failure but it's strictly defensive against a redis-py contract that already covers this. Open to changing if you have a repro of a leaked pool.

[coderabbitai]

@shacharm2, your reasoning holds up well. By default, an internal connection pool is created on Redis() (and Redis.from_url) and attached to the Redis instance lazily, so no socket is established before the PING is issued. When PING fails mid-handshake, redis-py's retry path calls _disconnect_raise_connect → connection.disconnect() before re-raising, so the half-open socket is released.

The one nuance worth flagging: using asyncio Redis requires an explicit disconnect of the connection since there is no asyncio destructor magic method. If PING raises and the caller doesn't hold a reference to client, the pool object lingers until GC and redis-py emits an "Unclosed Redis client" ResourceWarning. That's a hygiene/noise issue rather than a real resource leak, though — since the pool has no open connections at that point, the actual FD impact is zero.

Given that the PING failure path here is a fatal boot error (the process won't start), the ResourceWarning window is effectively zero between the raise and process exit. Your rebuttal is sound, and I'll withdraw the suggestion.

---

✏️ Learnings added

Learnt from: shacharm2
Repo: shacharm2/worthless PR: 71
File: src/worthless/proxy/metering.py:163-173
Timestamp: 2026-04-25T09:41:14.292Z
Learning: In `src/worthless/proxy/metering.py`, the maintainer deliberately chose NOT to wrap `await client.ping()` in `try/except + await client.aclose()` in `create_redis_client`. The reasoning: `Redis.from_url` creates a lazy pool (no upfront connections), PING is the first command, and redis-py's `disconnect_on_error` / retry path calls `connection.disconnect()` before re-raising on failure. A PING failure is a fatal boot error (process startup aborts), so any ResourceWarning from GC is inconsequential. The `aclose()` guard is considered overly defensive against a redis-py contract that already handles cleanup. Do not re-flag this pattern.

Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# 1) Confirm pytest collects this file at all — default python_files pattern is test_*.py.
rg -n '^python_files|^testpaths' pyproject.toml
# 2) Is there a CI or tox entry that overrides python_files for benchmarks?
rg -nC2 'bench_spend_cap_rule|benchmark' -g '!**/*.md' -g '!**/*.lock'
# 3) Check whether redis-py 5.x AsyncRedis is documented as loop-agnostic.
rg -nC3 'loop' -g '**/redis/asyncio/**' 2>/dev/null || true

Repository: shacharm2/worthless

Length of output: 10420

---

🌐 Web query:

redis-py AsyncRedis event loop binding across loops async connection pool

💡 Result:

In redis-py (redis.asyncio.Redis, formerly known as AsyncRedis), async connection pools and individual connections are bound to the event loop where they are first used. Connections use asyncio.open_connection, which creates asyncio.StreamReader/Writer tied to the running event loop via asyncio.get_running_loop or similar calls during connection establishment. Futures and tasks created for I/O operations are loop-specific. Key findings: - Sharing a single ConnectionPool or Redis client across multiple event loops causes RuntimeError: "got Future attached to a different loop" because asyncio primitives (sockets, futures) cannot cross loops. This is evident in GitHub issues #3351 and #3492, where shared pools fail in testing frameworks like pytest-asyncio (with per-test loops) or concurrent requests creating separate loops. - Official docs recommend creating pools/clients per event loop context. For shared access, use per-request dependencies in frameworks like FastAPI (create new Redis.from_pool(pool) or Redis per request, close after). - No explicit "loop binding" parameter exists; binding happens implicitly on first connect. Single-connection mode (single_connection_client=True) uses locks but still loop-bound. - Best practice: In single-loop apps (e.g., one asyncio.run), share one pool/client safely. In multi-loop scenarios (tests, workers), create new instances per loop. Use BlockingConnectionPool for waiting on connections without errors when pool exhausted. - Explicit close via await client.aclose or pool.aclose required, as no aexit magic for pools shared across clients. Examples from docs: pool = redis.asyncio.ConnectionPool.from_url("redis://localhost") client1 = redis.Redis(connection_pool=pool) # Multiple clients can share if same loop client2 = redis.Redis(connection_pool=pool) await client1.aclose await client2.aclose await pool.aclose For FastAPI dependency (avoids multi-loop issues): async def get_redis: redis_con = redis.asyncio.Redis try: yield redis_con finally: await redis_con.aclose

Citations:

• 1: https://redis.readthedocs.io/en/stable/examples/asyncio_examples.html
• 2: https://github.com/redis/redis-py/blob/master/redis/asyncio/client.py
• 3: https://redis.readthedocs.io/en/v7.0.1/examples/asyncio_examples.html
• 4: Error when reusing asyncio connection pool - multiple event loops redis/redis-py#3351
• 5: https://redis.readthedocs.io/en/stable/connections.html
• 6: https://redis.readthedocs.io/en/stable/_modules/redis/asyncio/connection.html
• 7: https://github.com/redis/redis-py/blob/master/redis/asyncio/connection.py

---

🏁 Script executed:

#!/bin/bash
# Find the docker_required decorator definition
rg -n 'def docker_required|@docker_required' tests/bench_spend_cap_rule.py -A 3 | head -20
rg -n 'def docker_required' tests/ -A 10

Repository: shacharm2/worthless

Length of output: 320

---

🏁 Script executed:

#!/bin/bash
# Check if docker_required is a skip decorator or something else
rg -n 'docker_required' tests/conftest.py -A 5 -B 2

Repository: shacharm2/worthless

Length of output: 45

---

🏁 Script executed:

#!/bin/bash
# Check CI configuration to see if TCP tests are run
fd -type f -name '*.yml' -o -name '*.yaml' -o -name '.github' | head -10
git ls-files '.github/**' 2>/dev/null | head -20

Repository: shacharm2/worthless

Length of output: 647

---

🏁 Script executed:

#!/bin/bash
# Find docker_required definition across all files
rg -n 'def docker_required|docker_required =' tests/ src/ -B 2 -A 5

Repository: shacharm2/worthless

Length of output: 1042

---

🏁 Script executed:

#!/bin/bash
# Check the benchmarks workflow
cat .github/workflows/benchmarks.yml

Repository: shacharm2/worthless

Length of output: 691

---

🏁 Script executed:

#!/bin/bash
# Check _docker_available() function to understand skip condition
rg -n '_docker_available' tests/bench_spend_cap_rule.py -B 2 -A 5

Repository: shacharm2/worthless

Length of output: 603

---

🏁 Script executed:

#!/bin/bash
# Check if there are any test runs or CI that specifically enable docker or set WORTHLESS_TEST_REDIS_URL
rg -n 'WORTHLESS_TEST_REDIS_URL|docker' .github/workflows/ -i

Repository: shacharm2/worthless

Length of output: 4526

---

🏁 Script executed:

#!/bin/bash
# Look at the actual benchmark test implementations to understand the loop issue
sed -n '361,402p' tests/bench_spend_cap_rule.py

Repository: shacharm2/worthless

Length of output: 1450

---

🏁 Script executed:

#!/bin/bash
# Confirm GitHub Actions ubuntu-latest has docker by default
# Check if there's explicit docker setup in benchmarks workflow
cat .github/workflows/benchmarks.yml | grep -A 5 -B 5 "docker\|service"

Repository: shacharm2/worthless

Length of output: 45

---

🏁 Script executed:

#!/bin/bash
# Check if seeded_tcp is used anywhere else or if there are other similar patterns
rg -n 'seeded_tcp|create_redis_client' tests/bench_spend_cap_rule.py | head -20

Repository: shacharm2/worthless

Length of output: 548

---

🏁 Script executed:

#!/bin/bash
# Final verification: check what ubuntu-latest provides for docker
# Also confirm the exact fixture loop binding by looking at tcp_redis_url fixture
rg -n 'tcp_redis_url|@pytest.fixture' tests/bench_spend_cap_rule.py | head -20

Repository: shacharm2/worthless

Length of output: 381

---

🏁 Script executed:

#!/bin/bash
# Verify the create_redis_client function to confirm loop binding happens
rg -n 'async def create_redis_client' worthless/ -A 10

Repository: shacharm2/worthless

Length of output: 114

---

🏁 Script executed:

#!/bin/bash
# Find the correct path for create_redis_client
find . -name '*.py' -type f | xargs rg -l 'def create_redis_client' | head -5

Repository: shacharm2/worthless

Length of output: 97

---

🏁 Script executed:

#!/bin/bash
# Check the implementation
rg -n 'async def create_redis_client' -A 15

Repository: shacharm2/worthless

Length of output: 1293

---

🏁 Script executed:

#!/bin/bash
# Double check: does GitHub Actions ubuntu-latest have docker as a default tool?
# Look for any docker setup steps or environment info in CI files
cat .github/workflows/benchmarks.yml

Repository: shacharm2/worthless

Length of output: 691

---

🏁 Script executed:

#!/bin/bash
# Final check: look at the full seeded_tcp fixture to confirm the loop binding
sed -n '320,360p' tests/bench_spend_cap_rule.py

Repository: shacharm2/worthless

Length of output: 1625

---

Cross-loop usage of a real redis-py client will fail in the benchmarks workflow — the TCP tests do run and will hit the "Task got Future attached to a different loop" error.

The async seeded_tcp fixture (line 321) creates the Redis client via await create_redis_client(...) on pytest-asyncio's event loop, binding the client and its connection pool to that loop. Each benchmark test then spins up its own asyncio.new_event_loop() (lines 371, 395) and drives rule.evaluate() on it — causing redis-py's async pool to attempt operations from a different loop than the one where its connections were created.

The @docker_required decorator will not skip these tests in the benchmarks workflow because GitHub Actions ubuntu-latest includes Docker by default. When the benchmarks run, this pattern will surface the documented redis-py error (see issues #3351, #3492).

The comment at lines 346–349 acknowledges the teardown side of the same issue, but the benchmark body itself hits it first. Fix by making the fixture lazy/sync so the client is created inside the benchmark's own loop:

Example approach (lines 320–350)

`@pytest.fixture`(params=LEDGER_SIZES, ids=lambda n: f"ledger={n}")
def seeded_tcp(tmp_path, request, tcp_redis_url):
    """Same seed pattern as `seeded`, but wires a real TCP Redis client."""
    db_path = tmp_path / "worthless.db"
    # Initialize database synchronously or defer redis creation to test
    def _setup_redis():
        redis = asyncio.run(create_redis_client(tcp_redis_url))
        return redis
    
    yield _setup_redis, "alice", request.param

Or refactor tests to call a setup function inside their own event loop before benchmark runs.

Also applies to: 361–402

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/bench_spend_cap_rule.py` around lines 320 - 358, The seeded_tcp fixture
binds a redis-py async client to pytest-asyncio's loop causing "Future attached
to a different loop" when benchmarks create their own event loop; change
seeded_tcp to be synchronous/lazy so the actual create_redis_client call happens
inside the benchmark's loop (i.e. convert async def seeded_tcp -> def seeded_tcp
and yield a factory like _make_redis that the test calls inside its event loop
to await create_redis_client(tcp_redis_url), then call incr_spend_hot(redis,
"alice", ...) and construct SpendCapRule(db=redis_db, redis=redis) there);
ensure the factory returns the same tuple (rule, "alice", size) or returns the
redis and spend_key/alice so the test can create SpendCapRule and perform proper
async cleanup (close redis with redis.aclose() and sqlite with redis_db.close())
from the benchmark loop.

[oblangatas]

Acknowledged. The benchmark is intentionally not part of the default test_* collection (benchmarks aren't regression tests; they go through pytest-benchmark's explicit invocation: pytest tests/bench_spend_cap_rule.py --benchmark-only ...). The file is collected when you point pytest at it directly, which is what the writeup at .planning/bench/spend-cap-rule-sqlite-vs-redis.md documents. Considering adding bench_*.py to python_files for safety, but that risks bench runs sneaking into the default suite. Leaving as-is for now.

[coderabbitai]

@shacharm2, understood — the bench_ prefix keeps these out of the default test_* collection, which is a clean separation between regression tests and benchmarking runs.

One note worth keeping in mind: the cross-loop concern I raised applies to the explicit benchmark run too (not just CI's default suite). When pytest tests/bench_spend_cap_rule.py --benchmark-only is executed, the seeded_tcp fixture still creates the real redis-py client in pytest-asyncio's event loop, while each benchmark iteration drives rule.evaluate() from a new_event_loop(). So the "Future attached to a different loop" error would surface during the TCP-Redis benchmark cases (test_bench_single_tcp_redis / test_bench_concurrent_tcp_redis) on any run that actually reaches those tests.

If this becomes a problem, the simplest fix would be to make seeded_tcp a synchronous fixture that yields a factory callable, deferring create_redis_client + SpendCapRule construction into each test's own loop — but that's entirely your call given the benchmarking context. Happy to leave it here.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

noeviction policy assertion couples to the container fixture and can spuriously fail against an external WORTHLESS_TEST_REDIS_URL.

docker_required skips only when both docker is unavailable and WORTHLESS_TEST_REDIS_URL is unset, so a developer who points WORTHLESS_TEST_REDIS_URL at their own Redis (e.g. a dev box with allkeys-lru) will hit this test and see a failure that reflects their local config, not the code under review. The in-code comment acknowledges this, but still recommends keeping it "green" — the assertion will turn red.

Consider gating this specific test on the container path (e.g. a sub-fixture flag) so pre-existing-URL runs skip it, or at minimum emitting a helpful pytest.skip message when the policy isn't noeviction and the URL came from the env.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_redis_metering_dynamic.py` around lines 542 - 555, Modify
test_real_redis_noeviction_policy_is_set so it does not fail when pointing at an
external Redis via WORTHLESS_TEST_REDIS_URL: after fetching policy from
real_redis_client (in test_real_redis_noeviction_policy_is_set), check
os.environ.get("WORTHLESS_TEST_REDIS_URL") (or a flag on the real_redis_client
fixture if available) and if the env var is set and the decoded value is not
"noeviction", call pytest.skip with a clear message; ensure pytest and os are
imported and leave the existing docker_required decorator and real_redis_client
usage unchanged.

[oblangatas]

Acknowledged but keeping. The test only runs under @docker_required and uses the session-scoped container fixture; the dev-convenience env override (WORTHLESS_TEST_REDIS_URL) is documented as 'use at your own risk' for a reason. If a developer points it at allkeys-lru and runs this single test, they get a test-config diagnostic, not a code regression. The CI canonical path (Docker container with our compose-managed config) is what matters and is what the test asserts. Adding a graceful skip on policy mismatch is reasonable but feels like papering over user error — open to a small follow-up if you disagree.

[coderabbitai]

@shacharm2, that's a fair call. The distinction between "our container, our policy" and "external Redis, your problem" is a reasonable line to draw, and the in-code comment already documents the intentional asymmetry. I'll leave it as-is.

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (3)

tests/test_redis_metering_failure_modes.py (1)

111-119: _DriftingRedis is defined but never used.

The class has no body beyond a docstring and no references elsewhere in the file — the drift test at lines 271-300 uses _FakeRedis directly and mutates r.store[...] at line 291 to force the stale value. Either delete _DriftingRedis or move the drift-simulation knob (e.g., an overridden get that always returns a planted stale value) into it and switch the xfail test to instantiate it — that would make the intent of the drift scenario self-documenting.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_redis_metering_failure_modes.py` around lines 111 - 119, The
_DriftingRedis class is declared but unused; either remove it or implement its
intended behavior and update the drift test to use it: modify _DriftingRedis
(subclassing _FakeRedis) to override get (or whatever lookup method _FakeRedis
uses) to return a planted stale value for the specific key, then change the
xfail drift test to instantiate _DriftingRedis instead of _FakeRedis and stop
directly mutating r.store; this makes the drift simulation self-documenting
while keeping _FakeRedis for other tests.

tests/test_redis_metering_properties.py (1)

390-510: Temp .db files leak across stateful examples.

_bootstrap creates the DB with NamedTemporaryFile(..., delete=False) (line 395) and teardown (lines 498-510) never unlinks self._db_path. Each Hypothesis example leaves one .db in the temp dir (up to max_examples=20 × any CI re-runs, plus any WAL/SHM siblings). Not catastrophic, but trivially avoidable and noisy on repeat local runs.

🧹 Suggested tweak

     def teardown(self) -> None:
         if self._loop is None:
             return
 
         async def _close() -> None:
             if self._db is not None:
                 await self._db.close()
 
         try:
             self._loop.run_until_complete(_close())
         finally:
             self._loop.close()
             self._loop = None
+            if self._db_path is not None:
+                import contextlib
+                from pathlib import Path
+                for suffix in ("", "-wal", "-shm"):
+                    with contextlib.suppress(FileNotFoundError):
+                        Path(self._db_path + suffix).unlink()

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_redis_metering_properties.py` around lines 390 - 510, The temp DB
path created in _bootstrap (tmp = NamedTemporaryFile(..., delete=False) ->
self._db_path) is never removed, leaking .db (and .db-shm/.db-wal) files; update
teardown to unlink self._db_path and any sibling WAL/SHM files after closing
self._db (inside the async _close or immediately after run_until_complete) using
os.unlink (guarded with exists/try/except to ignore missing files), and add the
os import; ensure cleanup runs in the finally block so files are removed even on
failures.

tests/conftest.py (1)

248-252: Prefer item.path (pathlib) over item.fspath in pytest 8+.

item.path (pathlib.Path) became the standard in pytest 8.0.0 and is the recommended attribute going forward. The legacy item.fspath still works, but aligns your code with modern pytest idioms since the project requires pytest≥8.0.

🧹 Suggested refactor

 def pytest_collection_modifyitems(config, items):  # noqa: ARG001 — pytest hook signature
     for item in items:
-        path = str(item.fspath)
+        path = str(item.path)
         if any(marker in path for marker in _REDIS_TEST_FILES):
             item.add_marker(pytest.mark.redis)

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/conftest.py` around lines 248 - 252, In pytest_collection_modifyitems,
replace the legacy item.fspath usage with the modern pathlib-based item.path:
change path = str(item.fspath) to path = str(item.path) (or use
item.path.as_posix() if you prefer POSIX strings) so the hook uses pytest≥8
idioms; keep the rest of the logic (checking _REDIS_TEST_FILES and adding
pytest.mark.redis) unchanged and ensure imports or type assumptions elsewhere
still accept a string or Path as appropriate.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@src/worthless/proxy/rules.py`:
- Around line 163-188: The TOCTOU comes from calling get_spend_hot and
rehydrate_spend_hot outside self._reserve_lock; to fix, re-read the Redis
counter (and if needed rehydrate) while holding self._reserve_lock before doing
the >= spend_cap check: keep the existing outer attempt to read
get_spend_hot/rehydrate_spend_hot for fast-path, but after entering async with
self._reserve_lock call get_spend_hot again (and if it returns None call
rehydrate_spend_hot) and use that fresh value (with the already_reserved from
self._reserved) for the capacity check and reservation; preserve existing
exception handling (fall back to _evaluate_sqlite or spend_cap_error_response)
and continue to reference the same symbols (_reserve_lock, get_spend_hot,
rehydrate_spend_hot, _reserved, record_spend, release_reservation,
_evaluate_sqlite).

---

Nitpick comments:
In `@tests/conftest.py`:
- Around line 248-252: In pytest_collection_modifyitems, replace the legacy
item.fspath usage with the modern pathlib-based item.path: change path =
str(item.fspath) to path = str(item.path) (or use item.path.as_posix() if you
prefer POSIX strings) so the hook uses pytest≥8 idioms; keep the rest of the
logic (checking _REDIS_TEST_FILES and adding pytest.mark.redis) unchanged and
ensure imports or type assumptions elsewhere still accept a string or Path as
appropriate.

In `@tests/test_redis_metering_failure_modes.py`:
- Around line 111-119: The _DriftingRedis class is declared but unused; either
remove it or implement its intended behavior and update the drift test to use
it: modify _DriftingRedis (subclassing _FakeRedis) to override get (or whatever
lookup method _FakeRedis uses) to return a planted stale value for the specific
key, then change the xfail drift test to instantiate _DriftingRedis instead of
_FakeRedis and stop directly mutating r.store; this makes the drift simulation
self-documenting while keeping _FakeRedis for other tests.

In `@tests/test_redis_metering_properties.py`:
- Around line 390-510: The temp DB path created in _bootstrap (tmp =
NamedTemporaryFile(..., delete=False) -> self._db_path) is never removed,
leaking .db (and .db-shm/.db-wal) files; update teardown to unlink self._db_path
and any sibling WAL/SHM files after closing self._db (inside the async _close or
immediately after run_until_complete) using os.unlink (guarded with
exists/try/except to ignore missing files), and add the os import; ensure
cleanup runs in the finally block so files are removed even on failures.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 2352aecf-1cce-4e65-a55e-0ddab711e689

📥 Commits

Reviewing files that changed from the base of the PR and between cf1fc2e and fecc5e0.

📒 Files selected for processing (5)

• pyproject.toml
• src/worthless/proxy/rules.py
• tests/conftest.py
• tests/test_redis_metering_failure_modes.py
• tests/test_redis_metering_properties.py

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

Narrow TOCTOU window between Redis GET and reservation lock.

get_spend_hot (line 164) and rehydrate_spend_hot (line 172) are called outside self._reserve_lock. If a concurrent background task completes record_spend (INCR on Redis) and release_reservation (decrement _reserved[alias]) between line 164 and line 179, this task reads a stale-low counter together with the already-decremented already_reserved. The sum counter + already_reserved then under-estimates true committed-plus-reserved by up to the released amount, potentially letting one request past the cap.

Contrast with _evaluate_sqlite, where the SELECT SUM(tokens) happens inside both the lock and BEGIN IMMEDIATE — genuinely race-free. The class docstring's claim that the reservation mechanism "eliminates the TOCTOU overrun" is therefore slightly stronger than what the Redis path actually delivers.

Impact is bounded (≤ one request's max_tokens overrun per hit, re-clamped on the next request since the SQLite ledger is authoritative and Redis INCR is monotonic), so this is a minor concern rather than a blocker. Options:

• Move get_spend_hot/rehydrate_spend_hot inside the lock, or
• Re-read the counter under the lock just before the >= spend_cap check, or
• Update the docstring to note the residual race on the Redis path.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/worthless/proxy/rules.py` around lines 163 - 188, The TOCTOU comes from
calling get_spend_hot and rehydrate_spend_hot outside self._reserve_lock; to
fix, re-read the Redis counter (and if needed rehydrate) while holding
self._reserve_lock before doing the >= spend_cap check: keep the existing outer
attempt to read get_spend_hot/rehydrate_spend_hot for fast-path, but after
entering async with self._reserve_lock call get_spend_hot again (and if it
returns None call rehydrate_spend_hot) and use that fresh value (with the
already_reserved from self._reserved) for the capacity check and reservation;
preserve existing exception handling (fall back to _evaluate_sqlite or
spend_cap_error_response) and continue to reference the same symbols
(_reserve_lock, get_spend_hot, rehydrate_spend_hot, _reserved, record_spend,
release_reservation, _evaluate_sqlite).

[oblangatas]

Acknowledged. The TOCTOU window between get_spend_hot and the reservation lock is real — _evaluate_redis doesn't have the same BEGIN IMMEDIATE serialization as _evaluate_sqlite. Impact bounded to ≤ one max_tokens overrun per concurrent burst (same class as the documented PoC limit in the SpendCapRule docstring, just with different wording). Tracked under worthless-dupf.2 (the broader provider-aware input+output reservation redesign), which would need a Lua/CAS reserve in Redis to fully close. Updating the docstring's 'eliminates the TOCTOU overrun' line to qualify it as 'eliminates the SQLite-path TOCTOU; Redis path bounded by the get→lock window' is a separate cleanup commit if you want it now.

[coderabbitai]

@shacharm2, thanks for the clear breakdown — the framing of "eliminates the SQLite-path TOCTOU; Redis path bounded by the get→lock window" is exactly the right distinction. Yes, the docstring cleanup is worth landing now even as a standalone commit — it sets accurate expectations for future contributors reading the class without having to chase the worthless-dupf.2 ticket. The Lua/CAS reserve (or a Redis EVAL-based check-and-reserve) is the right long-term closure, but that's clearly a bigger lift tied to the provider-aware reservation redesign.

Feel free to ship the docstring commit separately; no need to block this PR on it.

---

✏️ Learnings added

Learnt from: shacharm2
Repo: shacharm2/worthless PR: 71
File: src/worthless/proxy/rules.py:0-0
Timestamp: 2026-04-25T09:41:04.380Z
Learning: In `src/worthless/proxy/rules.py`, `SpendCapRule._evaluate_redis` has a residual TOCTOU window: `get_spend_hot`/`rehydrate_spend_hot` are called outside `self._reserve_lock`, so a concurrent `record_spend` INCR + `release_reservation` can cause the counter+reserved sum to under-estimate by up to one `max_tokens`. Impact is bounded to ≤ one `max_tokens` overrun per concurrent burst (next request self-heals via monotonic Redis INCR). Full fix requires a Lua/CAS reserve in Redis, tracked under worthless-dupf.2. A docstring clarification ("eliminates the SQLite-path TOCTOU; Redis path bounded by the get→lock window") is a pending cleanup commit.

Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.

---

🧠 Learnings used

Learnt from: shacharm2
Repo: shacharm2/worthless PR: 95
File: docs/planning/wor-276-recovery-final-plan.md:63-75
Timestamp: 2026-04-24T21:36:03.266Z
Learning: In the WOR-276 workstream (repo: shacharm2/worthless), `docs/planning/wor-276-recovery-final-plan.md` is a v1 historical RED-phase artifact. A v2 pivot simplified `safe_restore`'s signature — it no longer threads `skip_delta`, `original_user_arg`, or `_hook_before_replace` as the v1 plan anticipated. The authoritative contract lives in `docs/planning/wor-276-redesign-v2.md`, and the actual public surface is in `src/worthless/cli/safe_rewrite.py`. Do not flag the v1 doc's signature snippet as a discrepancy.

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@src/worthless/proxy/rules.py`:
- Around line 194-206: The RedisValueError branch in the handling logic (where
get_spend_hot raises RedisValueError and sets counter = None) must call
rehydrate_spend_hot with force=True so the corrupted Redis key is overwritten
instead of using SET NX; update the call to rehydrate_spend_hot(self.redis,
self.db, alias) to rehydrate_spend_hot(self.redis, self.db, alias, force=True)
(same behavior as the dirty-tracker path) and ensure this change is applied in
the method that currently calls get_spend_hot and rehydrate_spend_hot so
tampered/non-integer Redis values are replaced rather than left in Redis.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: d1f92679-9480-45d8-b20c-9f75940c6f85

📥 Commits

Reviewing files that changed from the base of the PR and between fecc5e0 and 359d74b.

📒 Files selected for processing (4)

• src/worthless/proxy/app.py
• src/worthless/proxy/metering.py
• src/worthless/proxy/rules.py
• tests/test_redis_metering_failure_modes.py

🚧 Files skipped from review as they are similar to previous changes (2)

• src/worthless/proxy/app.py
• tests/test_redis_metering_failure_modes.py

[coderabbitai]

Actionable comments posted: 5

🧹 Nitpick comments (7)

pyproject.toml (1)

197-202: Minor: S603 is already ignored globally, so the per-file additions are redundant.

S603 is in the top-level ignore list (Line 186), so listing it again in [tool.ruff.lint.per-file-ignores] for these test files has no effect. Only S607 is meaningfully needed here (matching the existing pattern at Line 196 for tests/test_docker_e2e.py). Consider dropping S603 from the three new entries to keep ignore semantics consistent across the file.

♻️ Proposed simplification

 # Redis dynamic tests shell out to `docker` CLI for the real-container tier
-"tests/test_redis_metering_dynamic.py" = ["S603", "S607"]
+"tests/test_redis_metering_dynamic.py" = ["S607"]
 # Benchmark harness shells out to `docker` CLI for the TCP-Redis backend
-"tests/bench_spend_cap_rule.py" = ["S603", "S607"]
+"tests/bench_spend_cap_rule.py" = ["S607"]
 # Streaming-Redis integration tests shell out to `docker` CLI for the real-container tier
-"tests/test_streaming_redis_integration.py" = ["S603", "S607"]
+"tests/test_streaming_redis_integration.py" = ["S607"]

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@pyproject.toml` around lines 197 - 202, The per-file ignores for
"tests/test_redis_metering_dynamic.py", "tests/bench_spend_cap_rule.py", and
"tests/test_streaming_redis_integration.py" redundantly include S603 which is
already in the top-level ignore list; edit the per-file entries to remove "S603"
and leave only "S607" so the per-file-ignores match the existing pattern and
avoid duplication of the globally ignored rule.

tests/test_redis_metering_defense_in_depth.py (2)

36-70: LGTM — coverage matches _ALIAS_RE semantics.

The reject cases (empty, colon, newline, traversal, space, non-string) line up cleanly with _ALIAS_RE.fullmatch(r"[a-zA-Z0-9_-]+") and the isinstance guard in spend_key.

One small tightening you may want: test_rejects_non_string allows either TypeError or ValueError, but the contract in spend_key() is specifically TypeError for a non-str. Pinning the test to TypeError would catch a regression where the type check is accidentally removed and the alias is fed straight into _ALIAS_RE.fullmatch, which would then raise the less-specific TypeError from re (or worse, succeed silently for some odd input).

Optional tightening

-    def test_rejects_non_string(self):
-        with pytest.raises((TypeError, ValueError)):
-            spend_key(123)  # type: ignore[arg-type]
+    def test_rejects_non_string(self):
+        with pytest.raises(TypeError, match="alias must be str"):
+            spend_key(123)  # type: ignore[arg-type]

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_redis_metering_defense_in_depth.py` around lines 36 - 70, Change
the non-string test to require a TypeError from spend_key so we enforce the
explicit isinstance guard: update
TestSpendKeyValidatesAlias.test_rejects_non_string to assert
pytest.raises(TypeError) when calling spend_key(123) (referencing the spend_key
function and its isinstance check) rather than allowing either TypeError or
ValueError, ensuring removal of the permissive `(TypeError, ValueError)` matcher
catches regressions that drop the type check.

---

85-100: Minor: prefer the public API over _dirty / _max_entries.

test_mark_beyond_cap_drops_oldest reaches into tracker._dirty to assert size, and test_default_cap_is_reasonable reads _max_entries. Both are fine for now, but the first one can be expressed via is_dirty so the test stays green if the internal container is ever swapped (e.g. OrderedDict → bespoke ring buffer). For example, asserting that alias-0 (the oldest) is no longer dirty after marking 50 with max_entries=10 exercises the ceiling without touching internals:

Optional refactor

     `@pytest.mark.asyncio`
     async def test_mark_beyond_cap_drops_oldest(self):
         tracker = SpendDirtyTracker(max_entries=10)
         for i in range(50):
             await tracker.mark(f"alias-{i}")
-        # Ceiling honoured.
-        assert len(tracker._dirty) <= 10
+        # Ceiling honoured: oldest must have been evicted, newest must remain.
+        assert not await tracker.is_dirty("alias-0")
+        assert await tracker.is_dirty("alias-49")

The _max_entries read in test_default_cap_is_reasonable is harder to avoid without exposing a property, so leaving that one as-is is reasonable.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_redis_metering_defense_in_depth.py` around lines 85 - 100, Replace
the internal-state assertions in test_mark_beyond_cap_drops_oldest to use
SpendDirtyTracker's public API: after creating tracker =
SpendDirtyTracker(max_entries=10) and marking alias-0..alias-49, drop the
len(tracker._dirty) assertion and instead assert that the oldest entry was
evicted via the public method, e.g. assert not tracker.is_dirty("alias-0") (and
you can optionally assert that a recent entry like "alias-49" is still dirty via
tracker.is_dirty("alias-49")); leave test_default_cap_is_reasonable unchanged
since _max_entries has no public accessor.

tests/test_streaming_redis_failure_modes.py (2)

298-344: Ordering risk: hardcoding rules[0] couples the test to fixture rule order.

app.state.rules_engine.rules[0] = SpendCapRule(...) assumes SpendCapRule is the first rule. The fixture happens to put it there, but the RulesEngine API treats rules as an ordered chain; reordering for any reason silently swaps the wrong rule. Cheap fix:

-    app.state.rules_engine.rules[0] = SpendCapRule(
-        db=app.state.db, redis=slow, dirty_tracker=app.state.dirty_tracker
-    )
-    # Re-local alias to the new rule for post-assertion.
-    rule = app.state.rules_engine.rules[0]
+    new_rule = SpendCapRule(
+        db=app.state.db, redis=slow, dirty_tracker=app.state.dirty_tracker
+    )
+    rules = app.state.rules_engine.rules
+    rules[:] = [new_rule if isinstance(r, SpendCapRule) else r for r in rules]
+    rule = new_rule

Also, _wait_for_background() is unused here in favor of the polling loop on slow.incr_calls — fine, but you could drop the helper call entirely or use it as a guard.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_streaming_redis_failure_modes.py` around lines 298 - 344, The test
mutates app.state.rules_engine.rules by hardcoding rules[0]; instead locate the
SpendCapRule instance instead of assuming index 0 (e.g., find the rule where
isinstance(rule, SpendCapRule) and replace that element in
app.state.rules_engine.rules), then rebind the local rule variable to that found
element; also either remove the unused _wait_for_background() helper or call it
as an additional guard after the request if you want its behavior. Ensure you
reference the rules list via app.state.rules_engine.rules and the SpendCapRule
class when implementing the lookup and replacement.

---

196-230: Verify that respx mocks httpx.AsyncClient calls dispatched by the proxy.

The fixture wires app.state.httpx_client = httpx.AsyncClient(follow_redirects=False) (default transport). @respx.mock patches the httpx.AsyncHTTPTransport.handle_async_request, which intercepts this client. Good. But other tests in this PR construct their own httpx.AsyncClient(transport=httpx.ASGITransport(app=app)) for the test client — that one talks to the ASGI app and must NOT be intercepted. Worth a one-liner sanity check on the first test that respx counts the upstream call exactly once, otherwise a future regression where the upstream client is built differently (e.g. mounting a custom transport) silently makes the assertion vacuous.

+    route = respx.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=httpx.ReadTimeout("simulated upstream readtimeout")
+    )
-    respx.post("https://api.openai.com/v1/chat/completions").mock(
-        side_effect=httpx.ReadTimeout("simulated upstream readtimeout")
-    )
@@
     await _wait_for_background()
+    assert route.called, "respx route was never hit; upstream may not be going through httpx_client"

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_streaming_redis_failure_modes.py` around lines 196 - 230, Add a
one-line sanity check in the test_upstream_readtimeout_releases_reservation test
to assert that respx intercepted exactly one upstream call: after installing the
respx.post mock
(respx.post("https://api.openai.com/v1/chat/completions").mock(...)) and after
making the client POST (or in the except branch), assert the respx route's call
count is 1 (use the respx route object or respx.calls/route.call_count) to
ensure app.state.httpx_client (the proxy's httpx.AsyncClient) was the client
actually exercised rather than some other AsyncClient/ASGITransport used by the
test client.

tests/test_streaming_redis_integration.py (2)

246-252: Bounded background polling — consider polling on a state predicate instead of fixed wall clock.

_wait_for_background busy-waits for up to ~500ms in 10ms ticks regardless of whether the BackgroundTask has flushed. On slow CI runners 500ms can be tight (the failure-modes file uses 800ms; inconsistency is itself a smell). Pattern in test_slow_redis_background_task_completes_bounded (poll until incr_calls >= 1) is more robust. Optional refactor, but tests that only assert "no row" (test 2) cannot poll on a positive signal and remain inherently racy. Consider raising the budget or hooking BackgroundTask completion via a sync primitive on app.state for deterministic waits.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_streaming_redis_integration.py` around lines 246 - 252, The helper
_wait_for_background should avoid fixed wall-clock busy-waiting; change it to
poll a real completion predicate or sync primitive instead of sleeping a fixed
500ms budget: either (A) accept a callable predicate and loop until predicate()
is true (similar to test_slow_redis_background_task_completes_bounded which
polls incr_calls >= 1) or (B) surface a completion Event/flag from
BackgroundTask on app.state (e.g., app.state.background_done) and await that
event here; also increase the default timeout if you keep a time budget. Update
references to _wait_for_background in tests to pass the appropriate predicate or
rely on app.state background completion so tests asserting "no row" are not
racy.

---

629-708: Tier-2 cleanup: redis.aclose() after cleanup() may double-close, and finally-block delete can swallow real bugs.

Two things:

1. _build_app_with_redis._cleanup already closes app.state.httpx_client and db but does NOT close app.state.redis. Here you await redis.aclose() after cleanup() — fine for this test, but the Tier-1 fixtures (redis_stack, redis_stack_anthropic) follow the same cleanup() → redis.aclose() ordering. Worth pulling redis closure into _cleanup so all three callers share one teardown.
2. try: await redis.delete(...) except Exception: pass will swallow a programming error (e.g. a typo in the spend_key after a refactor). Narrow it to transport/connection errors only:

-        try:
-            await redis.delete(spend_key(alias))
-        except Exception:
-            pass
+        try:
+            await redis.delete(spend_key(alias))
+        except RedisConnectionError:
+            pass

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_streaming_redis_integration.py` around lines 629 - 708, The test
currently closes Redis twice and swallows all exceptions on delete; update
_build_app_with_redis to include closing the Redis client in its returned
cleanup (ensure it closes app.state.redis / the redis object) so callers (this
test and Tier‑1 fixtures like redis_stack, redis_stack_anthropic) remove their
explicit await redis.aclose(), and in this test narrow the finally-block delete
to catch only transport/Redis errors (e.g., redis.exceptions.RedisError or
ConnectionError) instead of a blanket except Exception; reference the helper
_build_app_with_redis, its returned cleanup, the test's await redis.aclose(),
and spend_key(delete) call when making these changes.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In @.pre-commit-config.yaml:
- Around line 178-183: Update the advisory description for GHSA-58qw-9mgm-455v
in the comment: replace "tar-fallback symlink-extraction" with a concise
accurate phrase like "concatenated tar/ZIP interpretation conflict" while
preserving the rest of the context about toolchain-only, pip-audit -> pip-api ->
pip provenance and the note to revisit each release and remove the ignore when
the upstream fix lands.

In `@tests/test_streaming_redis_failure_modes.py`:
- Around line 277-287: Update the failing assertion message to accurately state
that the reservation release is performed by the BackgroundTask wrapping
_record_metering (not by the generator's finally), e.g. mention _record_metering
and BackgroundTask and that the generator's finally only calls
upstream_resp.aclose(); change the string that references "generator finally:
block" to reflect that the release is dispatched from
_record_metering/BackgroundTask so the assertion on rule._reserved and
redis.incr_calls is clear and correct.
- Around line 257-263: The tests pass raw bytes into httpx.Response(stream=...)
which causes async iteration failures; update each mocked response in
tests/test_streaming_redis_failure_modes.py (the respx.post(...) mocks around
the SSE tests) and the other listed test files to either use content=_SSE_PREFIX
(i.e. httpx.Response(200, content=_SSE_PREFIX, headers=...)) so httpx wraps it
as a ByteStream, or explicitly wrap the bytes as a ByteStream
(httpx.ByteStream(_SSE_PREFIX)) when using stream=; apply this change for the
mocks at the mentioned locations so adapter_resp.stream can be async-iterated
without RuntimeError.

In `@tests/test_streaming_redis_integration.py`:
- Around line 411-428: The test currently treats non-empty SQLite rows as
acceptable, which inverts the intended invariant: with the mock
`_OPENAI_SSE_NO_USAGE` the collector’s `result()` should be None and
`record_spend` must not run, so any rows from `_sqlite_spend_rows` indicate a
phantom spend and should fail loudly. Change the logic around `rows = await
_sqlite_spend_rows(...)` to assert that `rows` is empty (e.g., fail if `rows` is
truthy and include the `rows` contents in the message), and keep the existing
else branch checks that Redis `get_spend_hot(redis, alias)` is None and
`tracker.is_dirty(alias)` is False; also remove or update the misleading comment
and reference `StreamingUsageCollector.result()`, `_OPENAI_SSE_NO_USAGE`, and
`record_spend` in the test comment to reflect that non-empty `rows` is a test
failure.
- Around line 540-626: The real_redis_url fixture can leak containers if docker
port raises and should ensure cleanup and optional DB flush; move the
subprocess.run call that invokes ["docker", "port", name, "6379/tcp"]
(port_proc) inside the existing try block and call it with check=False so you
can inspect returncode and call pytest.skip/fail on failure while still hitting
the finally that stops the container (refer to variables name, run, port_proc,
ready); also when pre_existing from WORTHLESS_TEST_REDIS_URL is used, run a
FLUSHALL/FLUSHDB against that URL before yielding to avoid stale state (use the
same yield path as current) and add a docker label like "--label",
"worthless-test=1" to the docker run args so external janitors can prune
orphaned containers.

---

Nitpick comments:
In `@pyproject.toml`:
- Around line 197-202: The per-file ignores for
"tests/test_redis_metering_dynamic.py", "tests/bench_spend_cap_rule.py", and
"tests/test_streaming_redis_integration.py" redundantly include S603 which is
already in the top-level ignore list; edit the per-file entries to remove "S603"
and leave only "S607" so the per-file-ignores match the existing pattern and
avoid duplication of the globally ignored rule.

In `@tests/test_redis_metering_defense_in_depth.py`:
- Around line 36-70: Change the non-string test to require a TypeError from
spend_key so we enforce the explicit isinstance guard: update
TestSpendKeyValidatesAlias.test_rejects_non_string to assert
pytest.raises(TypeError) when calling spend_key(123) (referencing the spend_key
function and its isinstance check) rather than allowing either TypeError or
ValueError, ensuring removal of the permissive `(TypeError, ValueError)` matcher
catches regressions that drop the type check.
- Around line 85-100: Replace the internal-state assertions in
test_mark_beyond_cap_drops_oldest to use SpendDirtyTracker's public API: after
creating tracker = SpendDirtyTracker(max_entries=10) and marking
alias-0..alias-49, drop the len(tracker._dirty) assertion and instead assert
that the oldest entry was evicted via the public method, e.g. assert not
tracker.is_dirty("alias-0") (and you can optionally assert that a recent entry
like "alias-49" is still dirty via tracker.is_dirty("alias-49")); leave
test_default_cap_is_reasonable unchanged since _max_entries has no public
accessor.

In `@tests/test_streaming_redis_failure_modes.py`:
- Around line 298-344: The test mutates app.state.rules_engine.rules by
hardcoding rules[0]; instead locate the SpendCapRule instance instead of
assuming index 0 (e.g., find the rule where isinstance(rule, SpendCapRule) and
replace that element in app.state.rules_engine.rules), then rebind the local
rule variable to that found element; also either remove the unused
_wait_for_background() helper or call it as an additional guard after the
request if you want its behavior. Ensure you reference the rules list via
app.state.rules_engine.rules and the SpendCapRule class when implementing the
lookup and replacement.
- Around line 196-230: Add a one-line sanity check in the
test_upstream_readtimeout_releases_reservation test to assert that respx
intercepted exactly one upstream call: after installing the respx.post mock
(respx.post("https://api.openai.com/v1/chat/completions").mock(...)) and after
making the client POST (or in the except branch), assert the respx route's call
count is 1 (use the respx route object or respx.calls/route.call_count) to
ensure app.state.httpx_client (the proxy's httpx.AsyncClient) was the client
actually exercised rather than some other AsyncClient/ASGITransport used by the
test client.

In `@tests/test_streaming_redis_integration.py`:
- Around line 246-252: The helper _wait_for_background should avoid fixed
wall-clock busy-waiting; change it to poll a real completion predicate or sync
primitive instead of sleeping a fixed 500ms budget: either (A) accept a callable
predicate and loop until predicate() is true (similar to
test_slow_redis_background_task_completes_bounded which polls incr_calls >= 1)
or (B) surface a completion Event/flag from BackgroundTask on app.state (e.g.,
app.state.background_done) and await that event here; also increase the default
timeout if you keep a time budget. Update references to _wait_for_background in
tests to pass the appropriate predicate or rely on app.state background
completion so tests asserting "no row" are not racy.
- Around line 629-708: The test currently closes Redis twice and swallows all
exceptions on delete; update _build_app_with_redis to include closing the Redis
client in its returned cleanup (ensure it closes app.state.redis / the redis
object) so callers (this test and Tier‑1 fixtures like redis_stack,
redis_stack_anthropic) remove their explicit await redis.aclose(), and in this
test narrow the finally-block delete to catch only transport/Redis errors (e.g.,
redis.exceptions.RedisError or ConnectionError) instead of a blanket except
Exception; reference the helper _build_app_with_redis, its returned cleanup, the
test's await redis.aclose(), and spend_key(delete) call when making these
changes.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 7bb85ab1-3333-44e5-a601-988b3ef76b75

📥 Commits

Reviewing files that changed from the base of the PR and between 359d74b and 7dfb79e.

📒 Files selected for processing (6)

• .pre-commit-config.yaml
• pyproject.toml
• src/worthless/proxy/metering.py
• tests/test_redis_metering_defense_in_depth.py
• tests/test_streaming_redis_failure_modes.py
• tests/test_streaming_redis_integration.py

🚧 Files skipped from review as they are similar to previous changes (1)

• src/worthless/proxy/metering.py

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Verify how httpx and respx handle Response(stream=<bytes>):
# 1. Find the installed httpx Response.__init__ stream branch and aiter_bytes implementation.
# 2. Find respx's response factory to see if it wraps bytes.
fd -t f -e py 'pyproject.toml|requirements*.txt|uv.lock|poetry.lock' | head
rg -nP --type=py -C3 'httpx.Response\(\s*\d+\s*,\s*stream\s*=' tests | head -100

Repository: shacharm2/worthless

Length of output: 45

---

🌐 Web query:

In httpx 0.27+, does Response(status_code, stream=bytes_object, ...) iterate the bytes correctly via aiter_bytes, or does it require a SyncByteStream/AsyncByteStream subclass?

💡 Result:

In httpx 0.27+, a Response with stream=bytes_object iterates the bytes correctly via aiter_bytes without requiring a SyncByteStream/AsyncByteStream subclass. The ByteStream class wraps a bytes object and implements both iter (sync) and aiter (async), yielding the entire bytes object as a single chunk. The Response constructor accepts any SyncByteStream | AsyncByteStream | None for the stream argument, and since ByteStream(bytes_object) satisfies this (inheriting from both base classes), passing stream=bytes_object directly works if the type checker allows it, but explicitly using ByteStream(bytes_object) is safer and standard. For content=bytes_object, it is automatically converted to ByteStream via encode_response, which calls encode_content creating a ByteStream(body) with Content-Length header. In both cases, aiter_bytes will correctly iterate (yield) the bytes, typically as one chunk unless chunk_size is specified.

Citations:

• 1: https://github.com/encode/httpx/blob/master/httpx/_content.py
• 2: https://github.com/encode/httpx/blob/master/httpx/_models.py
• 3: https://github.com/encode/httpx/blob/0.28.1/httpx/_content.py
• 4: https://github.com/encode/httpx/blob/master/httpx/_types.py

---

🏁 Script executed:

# Find and read the actual test file
find . -name "test_streaming_redis_failure_modes.py" -type f

Repository: shacharm2/worthless

Length of output: 109

---

🏁 Script executed:

# Find project dependency files
fd -t f -e toml -e txt -e lock | grep -E "(pyproject|requirements|uv\.lock|poetry\.lock)"

Repository: shacharm2/worthless

Length of output: 86

---

🏁 Script executed:

# Check what _SSE_PREFIX is in the test file
rg -n "_SSE_PREFIX" tests/

Repository: shacharm2/worthless

Length of output: 297

---

🏁 Script executed:

# Read the test file around lines 257-263
sed -n '255,265p' ./tests/test_streaming_redis_failure_modes.py

Repository: shacharm2/worthless

Length of output: 443

---

🏁 Script executed:

# Read _SSE_PREFIX definition
sed -n '165,175p' ./tests/test_streaming_redis_failure_modes.py

Repository: shacharm2/worthless

Length of output: 450

---

🏁 Script executed:

# Find httpx and respx versions in dependencies
rg -A2 '"httpx|"respx' pyproject.toml

Repository: shacharm2/worthless

Length of output: 255

---

🏁 Script executed:

# Check if respx wraps bytes in stream parameter
# First find respx code or check its behavior
rg -n "stream" pyproject.toml | head -5

Repository: shacharm2/worthless

Length of output: 201

---

🏁 Script executed:

# Look at the test to understand respx usage more
sed -n '250,330p' ./tests/test_streaming_redis_failure_modes.py

Repository: shacharm2/worthless

Length of output: 3026

---

🏁 Script executed:

# Search for respx response mocking in tests to see if there's a pattern
rg -n "respx\.post.*stream=" tests/ | head -20

Repository: shacharm2/worthless

Length of output: 45

---

🏁 Script executed:

# Search for respx source to see if it auto-wraps stream parameter
# First, try to find respx in venv or site-packages
python3 << 'EOF'
import respx
import inspect
# Get the source of respx's Response or mock behavior
print(inspect.getfile(respx))
EOF

Repository: shacharm2/worthless

Length of output: 181

---

🏁 Script executed:

# Check what httpx.Response actually does with stream parameter
python3 << 'EOF'
import httpx
import inspect

# Check Response.__init__ signature
sig = inspect.signature(httpx.Response.__init__)
print("httpx.Response.__init__ signature:")
print(sig)

# Get source if available
try:
    source = inspect.getsource(httpx.Response.__init__)
    print("\nFirst 100 lines of httpx.Response.__init__:")
    for i, line in enumerate(source.split('\n')[:100], 1):
        print(f"{i:3}: {line}")
except:
    print("Could not get source")
EOF

Repository: shacharm2/worthless

Length of output: 2986

---

🏁 Script executed:

# Test if httpx.Response with raw bytes in stream parameter works
python3 << 'EOF'
import httpx
import asyncio

async def test_response_stream():
    # Try creating a Response with raw bytes in stream
    raw_bytes = b"data: test\n\n"
    
    try:
        resp = httpx.Response(200, stream=raw_bytes)
        print("Response created with stream=bytes successfully")
        print(f"resp.stream type: {type(resp.stream)}")
        
        # Try iterating
        chunks = []
        async for chunk in resp.aiter_bytes():
            chunks.append(chunk)
            print(f"Got chunk: {chunk!r}")
        print(f"Total chunks: {len(chunks)}")
    except Exception as e:
        print(f"Error: {type(e).__name__}: {e}")

asyncio.run(test_response_stream())
EOF

Repository: shacharm2/worthless

Length of output: 221

---

🏁 Script executed:

# Now verify what happens in the actual streaming code path
# Look for the code that iterates the response stream
rg -n "aiter_bytes\|aiter_raw" tests/test_streaming_redis_failure_modes.py

Repository: shacharm2/worthless

Length of output: 45

---

🏁 Script executed:

# Find the actual streaming consumer code mentioned in the review
rg -n "_stream_with_metering" src/ --type py

Repository: shacharm2/worthless

Length of output: 245

---

🏁 Script executed:

# Check if respx might be doing any wrapping when mocking
# Look at what respx.mock actually does
python3 << 'EOF'
import httpx

# Test if iterating raw bytes as sync works
raw_bytes = b"data: test\n\n"
try:
    for chunk in raw_bytes:
        print(f"Chunk from sync iter: {chunk!r} (type: {type(chunk).__name__})")
except Exception as e:
    print(f"Sync iter error: {e}")

# Now check what ByteStream does
from httpx._content import ByteStream
stream = ByteStream(raw_bytes)
print(f"\nByteStream wrapping:")
print(f"stream type: {type(stream)}")
async_gen = stream.__aiter__()
print(f"__aiter__ returns: {type(async_gen)}")
EOF

Repository: shacharm2/worthless

Length of output: 620

---

🏁 Script executed:

# Check the _stream_with_metering function to see how it consumes the stream
sed -n '469,520p' src/worthless/proxy/app.py

Repository: shacharm2/worthless

Length of output: 2583

---

🏁 Script executed:

# Look at where adapter_resp is created and how it's passed to _stream_with_metering
sed -n '450,480p' src/worthless/proxy/app.py

Repository: shacharm2/worthless

Length of output: 1662

---

🏁 Script executed:

# Verify if httpx.Response accepts a bytes object for stream parameter without type error
python3 << 'EOF'
import httpx

# Check if type checking allows bytes
raw_bytes = b"test"
resp = httpx.Response(200, stream=raw_bytes)
print(f"Created Response with stream=bytes: {type(resp.stream)}")

# Now try to actually use aiter_bytes() which is the real test
import asyncio

async def test_aiter():
    resp = httpx.Response(200, stream=raw_bytes)
    try:
        async for chunk in resp.aiter_bytes():
            print(f"Got chunk: {chunk!r}")
    except Exception as e:
        print(f"ERROR in aiter_bytes: {type(e).__name__}: {e}")
        import traceback
        traceback.print_exc()

asyncio.run(test_aiter())
EOF

Repository: shacharm2/worthless

Length of output: 678

---

🏁 Script executed:

# Check if there's a try/except or error handling in the test
sed -n '250,290p' tests/test_streaming_redis_failure_modes.py

Repository: shacharm2/worthless

Length of output: 1543

---

🏁 Script executed:

# Check lines 319-323 in the same file
sed -n '315,325p' tests/test_streaming_redis_failure_modes.py

Repository: shacharm2/worthless

Length of output: 363

---

🏁 Script executed:

# Check the test_streaming_redis_integration.py references at lines 339-343, 388-393, 445-451
sed -n '337,345p' tests/test_streaming_redis_integration.py
sed -n '386,395p' tests/test_streaming_redis_integration.py
sed -n '443,453p' tests/test_streaming_redis_integration.py

Repository: shacharm2/worthless

Length of output: 927

---

🏁 Script executed:

# Final confirmation: see if the test actually runs and what happens
# Check if there's a way respx might be intercepting and wrapping
python3 << 'EOF'
import httpx
from httpx._content import ByteStream

# Simulate what should happen
raw_bytes = b"data: test\n\n"

# Current broken approach (what the test does)
resp_broken = httpx.Response(200, stream=raw_bytes)
print(f"Broken approach - stream type: {type(resp_broken.stream)}")

# Correct approach 1: content parameter
resp_content = httpx.Response(200, content=raw_bytes)
print(f"Correct approach 1 (content=) - stream type: {type(resp_content.stream)}")

# Correct approach 2: explicit ByteStream
resp_wrapped = httpx.Response(200, stream=ByteStream(raw_bytes))
print(f"Correct approach 2 (ByteStream wrapper) - stream type: {type(resp_wrapped.stream)}")

# Now verify which ones work with aiter_bytes
import asyncio

async def test_approach(name, resp):
    print(f"\nTesting {name}:")
    try:
        chunks = []
        async for chunk in resp.aiter_bytes():
            chunks.append(chunk)
        print(f"  SUCCESS: Got {len(chunks)} chunk(s)")
    except Exception as e:
        print(f"  FAILED: {type(e).__name__}: {e}")

async def run_tests():
    await test_approach("raw bytes in stream=", resp_broken)
    await test_approach("content= parameter", resp_content)
    await test_approach("ByteStream wrapper", resp_wrapped)

asyncio.run(run_tests())
EOF

Repository: shacharm2/worthless

Length of output: 484

---

Fix httpx.Response(stream=<bytes>) to use proper byte stream wrapping across all test mocks.

The test mocks pass raw bytes to httpx.Response(stream=...), which breaks async iteration. httpx 0.28+ expects stream to be SyncByteStream | AsyncByteStream. When the production code calls async for chunk in adapter_resp.stream: (line 469 in src/worthless/proxy/app.py), it raises RuntimeError: Attempted to call an async iterator on a sync stream. The tests are exercising an error path, not the streaming success path they describe.

Fix by using content= instead: httpx.Response(200, content=_SSE_PREFIX, headers=...) — httpx automatically wraps it in a ByteStream. Alternatively, wrap explicitly: httpx.Response(200, stream=ByteStream(_SSE_PREFIX), ...).

Applies to:

• tests/test_streaming_redis_failure_modes.py lines 257–263, 319–323
• tests/test_streaming_redis_integration.py lines 339–343, 388–393, 445–451

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_streaming_redis_failure_modes.py` around lines 257 - 263, The
tests pass raw bytes into httpx.Response(stream=...) which causes async
iteration failures; update each mocked response in
tests/test_streaming_redis_failure_modes.py (the respx.post(...) mocks around
the SSE tests) and the other listed test files to either use content=_SSE_PREFIX
(i.e. httpx.Response(200, content=_SSE_PREFIX, headers=...)) so httpx wraps it
as a ByteStream, or explicitly wrap the bytes as a ByteStream
(httpx.ByteStream(_SSE_PREFIX)) when using stream=; apply this change for the
mocks at the mentioned locations so adapter_resp.stream can be async-iterated
without RuntimeError.

[oblangatas]

Per your own research note (the linked httpx docs check): httpx 0.27+ Response(stream=bytes_object) works — the bytes are wrapped in ByteStream which implements __iter__ + __aiter__ and yields the bytes as a single chunk. respx returns the response untouched. The 7 tests in this file all pass when run, including the three that use stream=<bytes>. Closing as not-an-issue.

[coderabbitai]

@shacharm2, you're right — I apologize for the incorrect flag. My sandbox environment ran an older httpx version that behaved differently, leading to a false positive. The httpx source confirms that Response.__init__ internally wraps a raw bytes argument in ByteStream (which implements both __iter__ and __aiter__), so stream=<bytes> is valid and async-iterable in httpx 0.27+. The tests passing confirms this. I'll record this for future reviews.

---

✏️ Learnings added

Learnt from: shacharm2
Repo: shacharm2/worthless PR: 71
File: tests/test_streaming_redis_failure_modes.py:257-263
Timestamp: 2026-04-25T09:41:09.445Z
Learning: In httpx 0.27+, passing raw `bytes` to `httpx.Response(stream=<bytes>)` is valid — httpx automatically wraps the bytes in an internal `ByteStream` that implements both `__iter__` and `__aiter__`, so the response can be async-iterated correctly (e.g. via `aiter_bytes()`). No explicit `ByteStream` wrapper or use of the `content=` parameter is required. Do not flag this pattern as a bug in test mocks using respx or httpx directly.

Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

Real-Redis fixture: handle docker port failures and avoid leaking stopped containers.

A few rough edges in the docker bring-up:

1. Lines 589-595: subprocess.run(..., check=True) with docker port. If it raises CalledProcessError, control jumps over the try: and the container started at line 559 is never stopped (the finally at line 623 is on the outer try, which we left via exception, but the exception propagates out of the fixture — pytest will see an error setting up the session, and the test session ends with an orphan container). Move docker port inside the existing try/finally and use check=False + pytest.skip/pytest.fail pattern you already use for redis-cli ping.
2. Line 622: when re-using WORTHLESS_TEST_REDIS_URL, the fixture yields and then returns — fine — but the FLUSHALL/FLUSHDB step is missing. Tests later delete(spend_key(alias)) per-alias (good), but a stale Redis DB inherited between sessions is still a footgun. Optional.
3. --rm plus an explicit docker stop is correct (stop triggers rm), but if the test process is SIGKILLed mid-session the container leaks. Consider --label worthless-test=1 so an external janitor can prune.

-    try:
-        port_proc = subprocess.run(  # noqa: S603, S607 — test-only
-            ["docker", "port", name, "6379/tcp"],
-            capture_output=True,
-            text=True,
-            timeout=5,
-            check=True,
-        )
-        host_port = port_proc.stdout.strip().split(":")[-1]
+    try:
+        port_proc = subprocess.run(  # noqa: S603, S607 — test-only
+            ["docker", "port", name, "6379/tcp"],
+            capture_output=True,
+            text=True,
+            timeout=5,
+            check=False,
+        )
+        if port_proc.returncode != 0 or not port_proc.stdout.strip():
+            pytest.skip(f"docker port lookup failed: {port_proc.stderr.strip()}")
+        host_port = port_proc.stdout.strip().split(":")[-1]

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	@pytest.fixture(scope="session")
	def real_redis_url():
	"""Return a URL to a real Redis daemon. Session-scoped: one container
	per test session.
	Order of preference:
	1. ``WORTHLESS_TEST_REDIS_URL`` env var - useful in CI or when the
	developer already has Redis running.
	2. Spin a ``redis:7-alpine`` container via ``docker run``.
	"""
	pre_existing = os.environ.get("WORTHLESS_TEST_REDIS_URL")
	if pre_existing:
	yield pre_existing
	return
	if not shutil.which("docker"):
	pytest.skip("docker not available and WORTHLESS_TEST_REDIS_URL unset")
	name = f"worthless-test-redis-{uuid.uuid4().hex[:8]}"
	run = subprocess.run( # noqa: S603, S607 — test-only
	[
	"docker",
	"run",
	"-d",
	"--rm",
	"--name",
	name,
	"-p",
	"127.0.0.1:0:6379",
	"redis:7-alpine",
	"redis-server",
	"--save",
	"",
	"--appendonly",
	"no",
	"--maxmemory",
	"32mb",
	"--maxmemory-policy",
	"noeviction",
	],
	capture_output=True,
	text=True,
	timeout=30,
	check=False,
	)
	if run.returncode != 0:
	pytest.skip(f"docker run failed: {run.stderr.strip()}")
	try:
	port_proc = subprocess.run( # noqa: S603, S607 — test-only
	["docker", "port", name, "6379/tcp"],
	capture_output=True,
	text=True,
	timeout=5,
	check=True,
	)
	host_port = port_proc.stdout.strip().split(":")[-1]
	url = f"redis://127.0.0.1:{host_port}/0"
	deadline = time.time() + 5.0
	last_err = ""
	ready = False
	while time.time() < deadline:
	probe = subprocess.run( # noqa: S603, S607 — test-only
	["docker", "exec", name, "redis-cli", "ping"],
	capture_output=True,
	text=True,
	timeout=2,
	check=False,
	)
	if probe.returncode == 0 and "PONG" in probe.stdout:
	ready = True
	break
	last_err = probe.stderr or probe.stdout
	time.sleep(0.2)
	if not ready:
	subprocess.run( # noqa: S603, S607 — test-only
	["docker", "stop", name], capture_output=True, timeout=15, check=False
	)
	pytest.skip(f"real redis did not become ready in 5s: {last_err}")
	yield url
	finally:
	subprocess.run( # noqa: S603, S607 — test-only
	["docker", "stop", name], capture_output=True, timeout=15, check=False
	)
	`@pytest.fixture`(scope="session")
	def real_redis_url():
	"""Return a URL to a real Redis daemon. Session-scoped: one container
	per test session.
	Order of preference:
	1. ``WORTHLESS_TEST_REDIS_URL`` env var - useful in CI or when the
	developer already has Redis running.
	2. Spin a ``redis:7-alpine`` container via ``docker run``.
	"""
	pre_existing = os.environ.get("WORTHLESS_TEST_REDIS_URL")
	if pre_existing:
	yield pre_existing
	return
	if not shutil.which("docker"):
	pytest.skip("docker not available and WORTHLESS_TEST_REDIS_URL unset")
	name = f"worthless-test-redis-{uuid.uuid4().hex[:8]}"
	run = subprocess.run( # noqa: S603, S607 — test-only
	[
	"docker",
	"run",
	"-d",
	"--rm",
	"--name",
	name,
	"-p",
	"127.0.0.1:0:6379",
	"redis:7-alpine",
	"redis-server",
	"--save",
	"",
	"--appendonly",
	"no",
	"--maxmemory",
	"32mb",
	"--maxmemory-policy",
	"noeviction",
	],
	capture_output=True,
	text=True,
	timeout=30,
	check=False,
	)
	if run.returncode != 0:
	pytest.skip(f"docker run failed: {run.stderr.strip()}")
	try:
	port_proc = subprocess.run( # noqa: S603, S607 — test-only
	["docker", "port", name, "6379/tcp"],
	capture_output=True,
	text=True,
	timeout=5,
	check=False,
	)
	if port_proc.returncode != 0 or not port_proc.stdout.strip():
	pytest.skip(f"docker port lookup failed: {port_proc.stderr.strip()}")
	host_port = port_proc.stdout.strip().split(":")[-1]
	url = f"redis://127.0.0.1:{host_port}/0"
	deadline = time.time() + 5.0
	last_err = ""
	ready = False
	while time.time() < deadline:
	probe = subprocess.run( # noqa: S603, S607 — test-only
	["docker", "exec", name, "redis-cli", "ping"],
	capture_output=True,
	text=True,
	timeout=2,
	check=False,
	)
	if probe.returncode == 0 and "PONG" in probe.stdout:
	ready = True
	break
	last_err = probe.stderr or probe.stdout
	time.sleep(0.2)
	if not ready:
	subprocess.run( # noqa: S603, S607 — test-only
	["docker", "stop", name], capture_output=True, timeout=15, check=False
	)
	pytest.skip(f"real redis did not become ready in 5s: {last_err}")
	yield url
	finally:
	subprocess.run( # noqa: S603, S607 — test-only
	["docker", "stop", name], capture_output=True, timeout=15, check=False
	)

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tests/test_streaming_redis_integration.py` around lines 540 - 626, The
real_redis_url fixture can leak containers if docker port raises and should
ensure cleanup and optional DB flush; move the subprocess.run call that invokes
["docker", "port", name, "6379/tcp"] (port_proc) inside the existing try block
and call it with check=False so you can inspect returncode and call
pytest.skip/fail on failure while still hitting the finally that stops the
container (refer to variables name, run, port_proc, ready); also when
pre_existing from WORTHLESS_TEST_REDIS_URL is used, run a FLUSHALL/FLUSHDB
against that URL before yielding to avoid stale state (use the same yield path
as current) and add a docker label like "--label", "worthless-test=1" to the
docker run args so external janitors can prune orphaned containers.

[oblangatas]

Acknowledged, fair improvements but parking as a follow-up. The fixture is a benchmark/integration aid; if docker port raises, the test session aborts loudly with the orphan visible to docker ps -a (annoying but not silent). Adding a robust try/except + pre-yield FLUSHALL is the right shape but touches multiple fixtures (test_redis_metering_dynamic.py has the same pattern). Filing as a small follow-up rather than scattering the fix across this PR.

[coderabbitai]

@shacharm2, that's a reasonable call — the failure mode is loud rather than silent, and a consistent fix across test_redis_metering_dynamic.py and test_streaming_redis_integration.py makes more sense than a half-landed patch in this PR.

Would you like me to open a GitHub issue to track this? I can file it covering all three points:

1. Move docker port inside try/finally with check=False + pytest.skip to prevent container leaks on port-lookup failures.
2. Pre-yield FLUSHALL/FLUSHDB when WORTHLESS_TEST_REDIS_URL is reused to avoid stale state.
3. Add --label worthless-test=1 to docker run for external janitor pruning.

I'd apply it to both fixture files in one go.

[oblangatas]

Parking this PR for v1.2.

After expert review (research + architect + brutus + benchmark) and a fresh look at the v1.1 audience, the Redis hot-path is correct future work but a poor fit for v1.1 solo/small-team users:

• Single-request loopback on macOS Docker is slower than SQLite (TCP overhead beats the gate-side speedup)
• Concurrent burst speedup only matters at >10 concurrent on one alias, which v1.1 users don't hit
• Deep-ledger speedup only matters at >1k rows per alias, bounded by the existing 90-day prune (WOR-182)
• Operational cost is real: extra service, password generation, AUTH/TLS story, healthcheck quirks, ACL syntax footguns we found and fixed

The branch is preserved as-is for v1.2 multi-tenant work — code, tests, benchmark, ACL all stay green. Cherry-picking the one truly SQLite-applicable fix (_reserved dict bound) into #107 so v1.1 still gets the leak fix.

Closes worthless-c3ox, worthless-d1au, worthless-dhwe as deferred to v1.2 with this PR as the historical artifact.