Skip to content

Implementation — usage.recorded events (billing backstop)

Issue BacMR #178 (rag half). Companion: gateway PR #180 consumes the event.

What was built

  • app/events/publisher.py — UsagePublisher (lazy aio-pika connection to the shared bacmr.events topic exchange) and emit_usage_recorded(...), a best-effort helper that publishes usage.recorded { user_id, tokens, request_id, reason }. It no-ops when RABBITMQ_URL is unset, tokens <= 0, or request_id is missing, and never raises into the caller. close_usage_publisher() runs on FastAPI shutdown.
  • app/api/routers/chat.py — adopts the gateway's forwarded id (request_id = get_request_id() or uuid4()). agent_streaming_generator now finalizes and publishes via a guarded finalize_and_publish() helper, called on normal completion and in an except (GeneratorExit, asyncio.CancelledError) branch so a mid-stream client disconnect still bills for what was generated. Internal errors still do not finalize/bill.
  • app/api/routers/quiz.py — publishes usage.recorded {reason:'quiz'} after a successful generate (already adopted get_request_id()).

Why

The gateway meters from the sync response; a chat client disconnecting before the SSE usage event flushes drops the charge. The out-of-band event closes that leak. The gateway's wallet_charge_outbox dedupes on request_id, so the sync charge and this event can't double-count — which is why chat now adopts the forwarded X-Request-ID.

Verification

  • pytest — 246 passed. New: publisher guard/no-op/publish/error-swallow; chat generator emits once on completion, once on aclose() (disconnect), and not on internal error; quiz emits with reason:'quiz'. Updated one pre-existing streaming test whose aclose() now bills.
  • ruff check . clean.
  • mkdocs build --strict clean.
  • Not exercised against a live broker (frugal test env has none); covered by unit tests.