Skip to content

Plan — Emit usage.recorded events (billing backstop)

Issue: BacMR #178 (RabbitMQ gaps), rag half of the usage.recorded billing backstop. Companion: gateway PR #180 consumes usage.recorded → wallet_charge_outbox.

Problem

The gateway meters chat/quiz by reading tokens_used from the sync HTTP/SSE response. If a chat client disconnects before the SSE usage event flushes, the gateway never sees the tokens and the charge is lost — rag did the work for free.

Approach

rag publishes usage.recorded { user_id, tokens, request_id, reason } out-of-band on the shared bacmr.events topic exchange after generation. The gateway's wallet_charge_outbox dedupes on request_id, so the sync path and this event can't double-charge — whichever lands first wins. Correctness depends on rag emitting with the same request_id the gateway forwards as X-Request-ID.

Changes

  1. app/events/publisher.py (new) — a lazy UsagePublisher (aio-pika) over the broker rag already consumes from; declares bacmr.events (topic, durable) and publishes persistent usage.recorded. emit_usage_recorded(...) is best-effort: no-ops when RABBITMQ_URL is unset, when tokens <= 0, or when request_id is missing, and never raises into the caller. close_usage_publisher() for clean shutdown.
  2. app/api/routers/chat.py — adopt the forwarded id: request_id = get_request_id() or uuid4() (was always uuid4()). In agent_streaming_generator, finalize + publish in a except (GeneratorExit, asyncio.CancelledError) branch so a mid-stream client disconnect still bills for what was generated. Normal completion finalizes + publishes once (guarded); internal errors still do not finalize or bill (unchanged).
  3. app/api/routers/quiz.py — publish usage.recorded {reason:'quiz'} after a successful generate (already uses get_request_id()).
  4. app/main.py — close the publisher on shutdown.

Non-goals

Ingestion-side token costs (operational, not per-user). Removing the sync metering (kept).

Repo-policy note

AGENTS.md §1.5(2) cautions against designs that "need a message broker." rag already ships an aio-pika consumer; this adds a publisher on the same broker (no new dependency), consistent with the cross-repo TODO.md mandate. Flagged and approved in Phase 1.

Tests

pytest: publisher no-op/guard/publish/error-swallow paths; chat generator emits once on normal completion and once on aclose() (disconnect); internal error still doesn't finalize; quiz emits with reason:'quiz'. Plus request-id adoption.