feat(llm-gateway): report org credit-bucket spend on the usage endpoint (#71404)

## Problem

PostHog Code is cutting over to usage-based billing, and the desktop app wants to show real org spend ("used $12.40 of your $50 limit") in the titlebar and on the Plan & usage page. Today `GET /v1/usage/{product}` carries only per-user percentages against internal rate valves plus an `ai_credits.exhausted` boolean, so clients have nothing to render a dollar amount from. The numbers already exist in `organization.usage`, synced from billing.

## Changes

- `GET /api/projects/{id}/quota_limits/` now returns `usage` and `limit` per resource, read from the org's synced billing numbers. `usage` is the same `usage + todays_usage` sum the quota limiter compares against the limit. The `limited` boolean stays authoritative for gating (grace periods and refund offsets live only in the limiting decision). Both fields are null when billing hasn't synced the resource.
- The gateway's quota resolver parses those numbers, converts credit buckets to USD (1 credit = $0.01) and carries them through its Redis cache. Fail-open paths keep them as null, meaning unknown, never $0.
- `GET /v1/usage/{product}` exposes them as `ai_credits.used_usd` and `ai_credits.limit_usd`. Additive and optional, so old clients are unaffected.

The per-user burst/sustained buckets still expose only percentages (`test_response_has_no_usd_fields` continues to guard that). The valve limits stay internal; the org's own bucket spend is the user's billing data.

> [!NOTE]
> The wire change is additive on both hops. Old gateway against new Django ignores the new keys; new gateway against old Django reads null and reports unknown.

## How did you test this code?

- Gateway: full unit suite passes locally (1240 tests). New regressions covered that no existing test did: Django responses without numbers read as unknown rather than $0, USD numbers survive the Redis cache round trip (a one-sided cache would flash the numbers in and out between hit and miss), fail-open cache entries carry null numbers, and the usage endpoint forwards the resolver's numbers onto `ai_credits`.
- Django: extended `test_quota_limits.py` with a synced-org case (1500 + 200 credits of 2000 reports usage 1700, a synced-but-unlimited bucket reports a null limit, a never-synced resource reports nulls, not zeros). I could not run the Django suite in this sandbox; CI covers it. Ruff (pinned version) is clean.
- mypy on the gateway files adds no new errors (two pre-existing ones in `quota_resolver.py` are untouched).

## Automatic notifications

- [ ] Publish to changelog?
- [ ] Alert Sales and Marketing teams?

## 🤖 Agent context

**Autonomy:** Human-driven (agent-assisted)

Authored by Claude (PostHog Code cloud session) at Adam's request, as the backend half of the usage-based billing client work in [PostHog/code#3487](PostHog/code#3487), which wants to render org spend for both free-tier and subscribed orgs. Main decision: Django reports raw native units generically for every resource, and the credits-to-USD conversion lives in the gateway where the client contract is defined, rather than special-casing credit buckets in Django. Absent-means-unknown is preserved end to end so a resolver blip or unsynced org never renders as $0 spent.

---
*Created with [PostHog Code](https://posthog.com/code?ref=pr)*

Author: adboio

ee/api/quota_limits.py

@@ -27,6 +27,34 @@ class QuotaResourceLimitSerializer(serializers.Serializer):
     limited = serializers.BooleanField(
         help_text="True when the team is currently over its quota for this resource and limits are in effect.",
     )
+    usage = serializers.FloatField(
+        allow_null=True,
+        help_text=(
+            "Units of this resource the organization has used so far this billing period, in the "
+            "resource's native unit (credits for credit buckets). Null when billing hasn't synced "
+            "usage for the resource."
+        ),
+    )
+    limit = serializers.FloatField(
+        allow_null=True,
+        help_text="The organization's limit for this resource in the same unit. Null when unlimited or unknown.",
+    )
+
+
+def _resource_usage(summary: dict[str, Any]) -> float | None:
+    """usage + todays_usage, the sum the quota limiter compares against the limit.
+
+    None rather than 0 when billing has never synced the resource, so clients read
+    it as unknown, not "$0 spent". The `limited` boolean stays authoritative for
+    gating; grace periods and refund offsets live only in that limiting decision.
+    """
+    if not summary:
+        return None
+    usage = summary.get("usage")
+    todays_usage = summary.get("todays_usage")
+    if usage is None and todays_usage is None:
+        return None
+    return (usage or 0) + (todays_usage or 0)
 
 
 class QuotaLimitsResponseSerializer(serializers.Serializer):
@@ -62,16 +90,19 @@ class QuotaLimitsViewSet(TeamAndOrgViewSetMixin, viewsets.ViewSet):
         responses={200: QuotaLimitsResponseSerializer},
     )
     def list(self, request: Request, *args: Any, **kwargs: Any) -> Response:
-        limited = {
-            resource.value: {
+        org_usage = self.team.organization.usage or {}
+        limited = {}
+        for resource in QuotaResource:
+            summary = org_usage.get(resource.value) or {}
+            limited[resource.value] = {
                 "limited": is_team_limited(
                     self.team.api_token,
                     resource,
                     QuotaLimitingCaches.QUOTA_LIMITER_CACHE_KEY,
                 ),
+                "usage": _resource_usage(summary),
+                "limit": summary.get("limit"),
             }
-            for resource in QuotaResource
-        }
         return Response(
             QuotaLimitsResponseSerializer(
                 {

ee/api/test/test_quota_limits.py

@@ -53,7 +53,7 @@ def test_session_auth_returns_under_quota_when_team_not_limited(self) -> None:
         response = self.client.get(self._url())
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         data = response.json()
-        self.assertEqual(data["limited"]["ai_credits"], {"limited": False})
+        self.assertEqual(data["limited"]["ai_credits"], {"limited": False, "usage": None, "limit": None})
         # Org holds no billing-granted Code usage feature -> reads as not paying
         self.assertIs(data["code_usage_billing_active"], False)
 
@@ -71,19 +71,42 @@ def test_reports_code_usage_billing_state(self) -> None:
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertIs(response.json()["code_usage_billing_active"], True)
 
+    def test_reports_org_usage_and_limit_for_synced_resources(self) -> None:
+        # The LLM gateway forwards these to clients (PostHog Code renders
+        # "used $X of $Y"); usage mirrors the limiter's usage + todays_usage sum.
+        self.organization.usage = {
+            "period": ["2026-07-01T00:00:00Z", "2026-08-01T00:00:00Z"],
+            "posthog_code_credits": {"usage": 1500, "todays_usage": 200, "limit": 2000},
+            "ai_credits": {"usage": 50, "todays_usage": 0},
+            "signals_credits": {"usage": None, "todays_usage": None, "limit": 5000},
+        }
+        self.organization.save()
+
+        response = self.client.get(self._url())
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        limited = response.json()["limited"]
+        self.assertEqual(limited["posthog_code_credits"], {"limited": False, "usage": 1700, "limit": 2000})
+        # Synced but unlimited: usage without a limit.
+        self.assertEqual(limited["ai_credits"], {"limited": False, "usage": 50, "limit": None})
+        # Synced with a limit but null usage figures: unknown usage, not zero.
+        self.assertEqual(limited["signals_credits"], {"limited": False, "usage": None, "limit": 5000})
+        # Never synced: unknown, not zero.
+        self.assertEqual(limited["events"], {"limited": False, "usage": None, "limit": None})
+
     def test_returns_limited_when_team_is_over_quota(self) -> None:
         self._set_ai_credits_limit(self.team.api_token, 9_999_999_999)
 
         response = self.client.get(self._url())
         self.assertEqual(response.status_code, status.HTTP_200_OK)
-        self.assertEqual(response.json()["limited"]["ai_credits"], {"limited": True})
+        self.assertEqual(response.json()["limited"]["ai_credits"], {"limited": True, "usage": None, "limit": None})
 
     def test_returns_unlimited_when_limit_has_already_expired(self) -> None:
         self._set_ai_credits_limit(self.team.api_token, 1)  # epoch 1970
 
         response = self.client.get(self._url())
         self.assertEqual(response.status_code, status.HTTP_200_OK)
-        self.assertEqual(response.json()["limited"]["ai_credits"], {"limited": False})
+        self.assertEqual(response.json()["limited"]["ai_credits"], {"limited": False, "usage": None, "limit": None})
 
     def test_personal_api_key_auth_works(self) -> None:
         self.client.logout()
@@ -102,7 +125,7 @@ def test_personal_api_key_auth_works(self) -> None:
             headers={"authorization": f"Bearer {raw_key}"},
         )
         self.assertEqual(response.status_code, status.HTTP_200_OK)
-        self.assertEqual(response.json()["limited"]["ai_credits"], {"limited": True})
+        self.assertEqual(response.json()["limited"]["ai_credits"], {"limited": True, "usage": None, "limit": None})
 
     def test_user_not_in_teams_org_is_forbidden(self) -> None:
         other_org = Organization.objects.create(name="other-org")
@@ -177,5 +200,5 @@ def test_multi_team_user_gets_per_team_answers(self) -> None:
         resp_self = self.client.get(self._url())
         resp_other = self.client.get(self._url(other_team.pk))
 
-        self.assertEqual(resp_self.json()["limited"]["ai_credits"], {"limited": True})
-        self.assertEqual(resp_other.json()["limited"]["ai_credits"], {"limited": False})
+        self.assertEqual(resp_self.json()["limited"]["ai_credits"], {"limited": True, "usage": None, "limit": None})
+        self.assertEqual(resp_other.json()["limited"]["ai_credits"], {"limited": False, "usage": None, "limit": None})

services/llm-gateway/src/llm_gateway/api/usage.py

@@ -37,6 +37,10 @@ class CostLimitStatus(BaseModel):
 
 class AiCreditsStatus(BaseModel):
     exhausted: bool
+    # Org-level bucket spend this billing period. None means unknown (unsynced
+    # org, resolver fail-open) — clients must not render None as $0.
+    used_usd: float | None = None
+    limit_usd: float | None = None
 
 
 class UsageResponse(BaseModel):
@@ -139,7 +143,11 @@ async def get_usage(
         user_id=user.user_id,
         burst=burst_status,
         sustained=sustained_status,
-        ai_credits=AiCreditsStatus(exhausted=credits_exhausted),
+        ai_credits=AiCreditsStatus(
+            exhausted=credits_exhausted,
+            used_usd=quota_status.used_usd,
+            limit_usd=quota_status.limit_usd,
+        ),
         is_rate_limited=burst_status.exceeded or sustained_status.exceeded or credits_exhausted,
         is_pro=is_pro_plan(plan_info.plan_key),
         code_usage_subscribed=quota_status.code_usage_billing_active,

services/llm-gateway/src/llm_gateway/services/quota_resolver.py

@@ -59,10 +59,30 @@
 )
 
 
+# Credit buckets are denominated in billing credits priced at one cent each
+# (the billing product config for AI credits and `posthog_code_usage`).
+_USD_PER_CREDIT = 0.01
+
+
 @dataclass
 class QuotaResourceStatus:
     limited: bool
     code_usage_billing_active: bool = False
+    # The org's bucket spend and limit this billing period, from billing's
+    # synced numbers. None means unknown (unsynced org, fail-open) — never $0.
+    used_usd: float | None = None
+    limit_usd: float | None = None
+
+
+def _optional_number(value: object) -> float | None:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return None
+    return float(value)
+
+
+def _credits_to_usd(credits: object) -> float | None:
+    value = _optional_number(credits)
+    return None if value is None else round(value * _USD_PER_CREDIT, 2)
 
 
 class _TransientUpstreamError(Exception):
@@ -187,6 +207,8 @@ async def _fetch(self, resource_key: str, team_id: int, auth_header: str) -> tup
         return QuotaResourceStatus(
             limited=bool(resource.get("limited")),
             code_usage_billing_active=bool(data.get("code_usage_billing_active")),
+            used_usd=_credits_to_usd(resource.get("usage")),
+            limit_usd=_credits_to_usd(resource.get("limit")),
         ), self._cache_ttl
 
     async def _get_cached(self, resource_key: str, team_id: int) -> QuotaResourceStatus | None:
@@ -202,6 +224,8 @@ async def _get_cached(self, resource_key: str, team_id: int) -> QuotaResourceSta
                 # Entries written before this field existed read as False (capped)
                 # until the TTL turns them over.
                 code_usage_billing_active=bool(payload.get("code_usage_billing_active")),
+                used_usd=_optional_number(payload.get("used_usd")),
+                limit_usd=_optional_number(payload.get("limit_usd")),
             )
         except Exception:
             logger.debug("quota_cache_read_failed", resource=resource_key, team_id=team_id)
@@ -212,7 +236,12 @@ async def _set_cached(self, resource_key: str, team_id: int, status: QuotaResour
             return
         try:
             payload = json.dumps(
-                {"limited": status.limited, "code_usage_billing_active": status.code_usage_billing_active}
+                {
+                    "limited": status.limited,
+                    "code_usage_billing_active": status.code_usage_billing_active,
+                    "used_usd": status.used_usd,
+                    "limit_usd": status.limit_usd,
+                }
             )
             await self._redis.set(_redis_key(resource_key, team_id), payload, ex=ttl)
         except Exception:

services/llm-gateway/tests/test_quota_resolver.py

@@ -223,12 +223,66 @@ async def test_billing_flag_falls_back_to_last_known_value_on_fetch_failure(self
             assert json.loads(redis.store[_redis_key("ai_credits", 42)]) == {
                 "limited": False,
                 "code_usage_billing_active": True,
+                "used_usd": None,
+                "limit_usd": None,
             }
             assert redis.ttls[_redis_key("ai_credits", 42)] == _FAIL_OPEN_CACHE_TTL_SECONDS
         else:
             # 4xx is caller-specific and must not repopulate the shared entry.
             assert _redis_key("ai_credits", 42) not in redis.store
 
+    @pytest.mark.asyncio
+    async def test_parses_credit_numbers_into_usd(self) -> None:
+        # Django reports credit-bucket usage/limit in credits (1 credit = $0.01);
+        # clients get dollars.
+        http_client = _make_http_client(
+            _make_response(
+                200,
+                {"limited": {"posthog_code_credits": {"limited": False, "usage": 1234, "limit": 2000}}},
+            )
+        )
+        resolver = QuotaResolver(redis=None, http_client=http_client)
+
+        status = await resolver.get_resource_status("posthog_code_credits", team_id=42, auth_header="Bearer phx_test")
+
+        assert status.used_usd == 12.34
+        assert status.limit_usd == 20.0
+
+    @pytest.mark.asyncio
+    async def test_missing_usage_numbers_read_as_unknown(self) -> None:
+        # Old Django responses and unsynced orgs carry no numbers — clients must
+        # see unknown, never $0.
+        http_client = _make_http_client(
+            _make_response(
+                200,
+                {"limited": {"posthog_code_credits": {"limited": False, "usage": None, "limit": None}}},
+            )
+        )
+        resolver = QuotaResolver(redis=None, http_client=http_client)
+
+        status = await resolver.get_resource_status("posthog_code_credits", team_id=42, auth_header="Bearer phx_test")
+
+        assert status.used_usd is None
+        assert status.limit_usd is None
+
+    @pytest.mark.asyncio
+    async def test_usd_numbers_round_trip_through_the_cache(self) -> None:
+        redis = _FakeRedis()
+        http_client = _make_http_client(
+            _make_response(
+                200,
+                {"limited": {"posthog_code_credits": {"limited": False, "usage": 1234, "limit": 5000}}},
+            )
+        )
+        resolver = QuotaResolver(redis=redis, http_client=http_client)  # type: ignore[arg-type]
+
+        first = await resolver.get_resource_status("posthog_code_credits", team_id=42, auth_header="Bearer phx_test")
+        cached = await resolver.get_resource_status("posthog_code_credits", team_id=42, auth_header="Bearer phx_test")
+
+        assert (first.used_usd, first.limit_usd) == (12.34, 50.0)
+        assert (cached.used_usd, cached.limit_usd) == (12.34, 50.0)
+        assert http_client.get.await_count == 1
+
     @pytest.mark.asyncio
     async def test_fetches_and_parses_unlimited_response(self) -> None:
         http_client = _make_http_client(
@@ -328,7 +382,12 @@ async def test_writes_cache_on_miss(self) -> None:
         quota_writes = [c for c in redis.set.await_args_list if c.args[0] == _redis_key("ai_credits", 42)]
         assert len(quota_writes) == 1
         call = quota_writes[0]
-        assert json.loads(call.args[1]) == {"limited": True, "code_usage_billing_active": False}
+        assert json.loads(call.args[1]) == {
+            "limited": True,
+            "code_usage_billing_active": False,
+            "used_usd": None,
+            "limit_usd": None,
+        }
         # Successful fetches use the gateway settings default of 5 minutes.
         assert call.kwargs.get("ex") == 300
         assert redis.set.await_count == 2

services/llm-gateway/tests/test_usage.py

@@ -378,7 +378,7 @@ def test_credits_reflect_products_own_bucket(self, authenticated_usage_client: T
         )
         assert response.status_code == 200
         data = response.json()
-        assert data["ai_credits"] == {"exhausted": True}
+        assert data["ai_credits"] == {"exhausted": True, "used_usd": None, "limit_usd": None}
         assert data["is_rate_limited"] is True
         assert resolver_mock.call_args.args[0] == "posthog_code_credits"
 
@@ -404,7 +404,7 @@ def test_exhausted_bucket_reported_for_every_caller(
         )
         assert response.status_code == 200
         data = response.json()
-        assert data["ai_credits"] == {"exhausted": True}
+        assert data["ai_credits"] == {"exhausted": True, "used_usd": None, "limit_usd": None}
         assert data["is_rate_limited"] is True
 
     def test_ai_credits_reflects_resolver_for_billable_product(self, authenticated_usage_client: TestClient) -> None:
@@ -420,10 +420,27 @@ def test_ai_credits_reflects_resolver_for_billable_product(self, authenticated_u
         )
         assert response.status_code == 200
         data = response.json()
-        assert data["ai_credits"] == {"exhausted": True}
+        assert data["ai_credits"] == {"exhausted": True, "used_usd": None, "limit_usd": None}
         assert data["is_rate_limited"] is True
         assert resolver_mock.call_args.args[0] == "ai_credits"
 
+    def test_ai_credits_carries_org_spend_numbers(self, authenticated_usage_client: TestClient) -> None:
+        """PostHog Code renders "used $X of $Y" (titlebar, plans page) off these
+        numbers; None must stay None so clients read unknown, not $0."""
+        from llm_gateway.services.quota_resolver import QuotaResourceStatus
+
+        app = authenticated_usage_client.app
+        app.state.quota_resolver.get_resource_status = AsyncMock(
+            return_value=QuotaResourceStatus(limited=False, used_usd=12.4, limit_usd=50.0)
+        )
+
+        response = authenticated_usage_client.get(
+            "/v1/usage/posthog_code",
+            headers={"Authorization": "Bearer phx_test"},
+        )
+        assert response.status_code == 200
+        assert response.json()["ai_credits"] == {"exhausted": False, "used_usd": 12.4, "limit_usd": 50.0}
+
     @pytest.mark.parametrize("billing_active", [True, False])
     def test_code_usage_subscribed_reflects_billing_bit(
         self, authenticated_usage_client: TestClient, billing_active: bool