budgets 0.5.0: cap per token (no pooling), one consumption record (#120)

Per draft-hardt-aauth-budgets editor's copy (Sept 2026):
- InMemoryMeter caps each auth token at its own budget (§Aggregation);
  the (iss, sub, aud) key remains the per-person ledger for records and
  usage counters, never a second ceiling. 0.4 pooled a person's live
  grants, letting a jti spend past its own grant with the overflow
  unattributed to the sibling it came from.
- Refusals carry ONE budget_consumed record — the presented token's
  {jti, consumed} (§The Consumption Record). New consumed_record(key, jti).
- Breaking: remaining(key) → remaining(key, jti); resource_token_provider
  receives a single record (dict | None).
- Vectors regenerated; 80 tests.

Author: abay94

CHANGELOG.md

@@ -1,5 +1,29 @@
 # Changelog
 
+## 0.5.0
+
+**AAuth Budgets — per-token cap, one consumption record** (draft-hardt-aauth-budgets
+editor's copy, September 2026; issue #120):
+
+- **No more pooling.** `InMemoryMeter` caps each auth token at its own `budget`
+  (§Aggregation): committed consumption plus outstanding reservations against
+  the presented `jti` never exceed *its* grant, and a token never draws on a
+  sibling's allocation. The `(iss, sub, aud)` key survives as the per-person
+  **ledger** for records and usage counters — it is not a second ceiling.
+  0.4 pooled a person's live grants into one purse, which let a jti spend past
+  its own grant; the overflow, spent from a sibling's allocation, was never
+  attributed and the sibling's remainder later released as unspent. The
+  `required` member makes a fragmented agent's re-authorization a calculation,
+  so the purse bought nothing worth that.
+- **One record on the wire.** Budget refusals carry a single
+  `budget_consumed` object — `{"jti", "consumed"}` for the PRESENTED token —
+  per §The Consumption Record. New `InMemoryMeter.consumed_record(key, jti)`;
+  `consumed_records(key, jkt=…)` stays as the audit view.
+- **Breaking:** `InMemoryMeter.remaining(key)` → `remaining(key, jti)`;
+  `resource_token_provider(key, records: list)` → `(key, record: dict | None)`.
+  `InsufficientBudget.remaining` is now the presented token's balance.
+- Test vectors regenerated (`consumption_records` is one object).
+
 ## 0.4.0
 
 **AAuth Budgets — the August 20 editor's-copy additions** (allocation model,

README.md

@@ -116,8 +116,10 @@ to the control plane. The middleware does the whole resource-side checklist:
 verify the token against your pinned PS, atomically reserve → commit →
 release per request, answer with `AAuth-Budget`, and refuse exhausted
 envelopes with a `401` + `AAuth-Requirement` (optionally carrying your signed
-resource token with the agent's own consumption records — scoped to its key,
-so one agent never learns about a sibling's spending):
+resource token with **one** consumption record — the presented token's
+`{jti, consumed}` — so one agent never learns about a sibling's spending).
+The cap is **per token**: a jti never draws on a sibling's grant; the
+`(iss, sub, aud)` key is only the per-person ledger behind records and usage:
 
 ```python
 from regent_httpsig import HttpsigConfig, HttpsigVerifier, InMemoryMeter

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "regent-httpsig"
-version = "0.4.0"
+version = "0.5.0"
 description = "Verify and sign AI agent HTTP traffic in Python — RFC 9421 HTTP Message Signatures: Web Bot Auth (what OpenAI ships) and AAuth."
 readme = "README.md"
 license = "Apache-2.0"

src/regent_httpsig/budget.py

@@ -5,11 +5,23 @@
     "budget": { "amount": 2000000, "unit": "USD", "decimals": 6 }   # = $2.00
 
 and the resource meters every request against it: reserve the request's maximum
-cost atomically, serve, commit the actual cost, release the difference. The
-draft requires consumption to be aggregated atomically across all live auth
-tokens for the key ``(iss, sub, aud)`` (§14.4), so the meter pools the grants
-of a principal's live tokens and counts reservations + consumption against
-that pool.
+cost atomically, serve, commit the actual cost, release the difference.
+
+Two things are counted, against different keys (draft §Aggregation):
+
+* the **cap** is per auth token — committed consumption plus outstanding
+  reservations against the presented ``jti`` never exceed *its* ``budget``;
+  no cross-token arithmetic, one token never draws on a sibling's grant;
+* the **ledger** is per person — consumption is posted to ``(iss, sub, aud)``
+  for the consumption record and the usage counters. It is not a second
+  ceiling: a request that fits its token's budget is never refused because
+  of a per-person total.
+
+(0.4 and earlier pooled a principal's live grants into one purse. That let a
+jti spend past its own grant, so the figure recorded against it could exceed
+what its issuer granted and the overflow — spent from a sibling's allocation —
+was never attributed. Dropped in 0.5.0; the ``required`` member makes a
+fragmented agent's re-authorization a calculation instead.)
 
 Everything here is framework-free; the FastAPI glue lives in
 :mod:`regent_httpsig.fastapi` (``BudgetMiddleware``).
@@ -42,7 +54,7 @@ class InvalidBudgetClaim(ValueError):
 
 
 class UnitMismatch(ValueError):
-    """A grant's unit/decimals differ from the pool's — one envelope, one unit."""
+    """A grant's unit/decimals differ from the ledger's — one envelope, one unit."""
 
 
 @dataclass(frozen=True)
@@ -80,8 +92,8 @@ def parse(claims: Mapping[str, Any]) -> BudgetClaim | None:
 
 @dataclass(frozen=True)
 class Reservation:
-    """An atomic hold on the pool for one in-flight request. Never revised —
-    committed (with the actual cost) or released, exactly once."""
+    """An atomic hold on one token's budget for one in-flight request. Never
+    revised — committed (with the actual cost) or released, exactly once."""
 
     rid: int
     key: MeterKey
@@ -91,7 +103,8 @@ class Reservation:
 
 @dataclass(frozen=True)
 class InsufficientBudget:
-    """Refusal: the request's maximum cost exceeds the pool's remaining balance.
+    """Refusal: the request's maximum cost exceeds the presented token's
+    remaining balance.
     ``exhausted`` distinguishes the draft's two reason tokens: an empty envelope
     (``budget-exhausted``) vs a too-expensive request (``insufficient-budget``)."""
 
@@ -204,21 +217,25 @@ def _purge(self, key: MeterKey, now: float) -> _Pool | None:
         return pool
 
     @staticmethod
-    def _remaining(pool: _Pool) -> int:
-        live = sum(a for a, _ in pool.grants.values())
-        spent = sum(pool.consumed.get(jti, 0) for jti in pool.grants)
-        held = sum(a for _, a, _ in pool.reservations.values())
-        return max(0, live - spent - held)
+    def _remaining(pool: _Pool, jti: str) -> int:
+        """The presented token's balance: its grant minus what was committed
+        against it minus what is held for it. Sibling tokens of the same
+        person do not enter — the cap is per token (§Aggregation)."""
+        grant = pool.grants.get(jti)
+        if grant is None:
+            return 0
+        held = sum(a for j, a, _ in pool.reservations.values() if j == jti)
+        return max(0, grant[0] - pool.consumed.get(jti, 0) - held)
 
     # ── public interface (the BudgetMeter contract) ──────────────────────────
 
     async def observe_grant(self, key: MeterKey, jti: str, claim: BudgetClaim,
                             exp: float, jkt: str = "") -> None:
-        """Register a token's envelope in the principal's pool (idempotent per
-        ``jti``). ``jkt`` is the RFC 7638 thumbprint of the token's ``cnf`` key —
+        """Register a token's envelope under the person's ledger key (idempotent
+        per ``jti``). ``jkt`` is the RFC 7638 thumbprint of the token's ``cnf`` key —
         recorded so consumption records can be scoped to the presenting agent
         (one agent must not learn about its siblings). Raises
-        :class:`UnitMismatch` if the pool already runs in a different unit —
+        :class:`UnitMismatch` if the ledger already runs in a different unit —
         one envelope, one unit, no FX at the meter."""
         async with self._lock:
             now = time.monotonic()
@@ -229,7 +246,7 @@ async def observe_grant(self, key: MeterKey, jti: str, claim: BudgetClaim,
                     key, _Pool(unit=claim.unit, decimals=claim.decimals))
             if (pool.unit, pool.decimals) != (claim.unit, claim.decimals):
                 raise UnitMismatch(
-                    f"pool runs in {pool.unit}/{pool.decimals}, "
+                    f"ledger runs in {pool.unit}/{pool.decimals}, "
                     f"grant is {claim.unit}/{claim.decimals}")
             pool.last_activity = now
             if jkt:
@@ -244,7 +261,7 @@ async def reserve(self, key: MeterKey, jti: str,
             pool = self._purge(key, now)
             if pool is None or jti not in pool.grants:
                 return InsufficientBudget(remaining=0, exhausted=True)
-            remaining = self._remaining(pool)
+            remaining = self._remaining(pool, jti)
             if max_cost > remaining:
                 return InsufficientBudget(remaining=remaining,
                                           exhausted=remaining == 0)
@@ -255,7 +272,7 @@ async def reserve(self, key: MeterKey, jti: str,
 
     async def commit(self, res: Reservation, actual: int) -> int:
         """Commit the actual cost (clamped to the reserved amount — reservations
-        are never revised upward) and return the pool's remaining balance."""
+        are never revised upward) and return the token's remaining balance."""
         async with self._lock:
             now = time.monotonic()
             pool = self._purge(res.key, now)
@@ -267,20 +284,22 @@ async def commit(self, res: Reservation, actual: int) -> int:
             pool.last_activity = now
             if cost > 0:
                 self._record_usage(res.key, pool, res.jti, cost)
-            return self._remaining(pool)
+            return self._remaining(pool, res.jti)
 
     async def release(self, res: Reservation) -> int:
         async with self._lock:
             pool = self._purge(res.key, time.monotonic())
             if pool is None:
                 return 0
             pool.reservations.pop(res.rid, None)
-            return self._remaining(pool)
+            return self._remaining(pool, res.jti)
 
-    async def remaining(self, key: MeterKey) -> int:
+    async def remaining(self, key: MeterKey, jti: str) -> int:
+        """The presented token's remaining balance (0 for an unknown or
+        expired ``jti``)."""
         async with self._lock:
             pool = self._purge(key, time.monotonic())
-            return 0 if pool is None else self._remaining(pool)
+            return 0 if pool is None else self._remaining(pool, jti)
 
     def _record_usage(self, key: MeterKey, pool: _Pool, jti: str,
                       amount: int) -> None:
@@ -322,11 +341,25 @@ def metering_unit(self) -> tuple[str, int] | None:
         or ``None`` before the first commit."""
         return self._metering_unit
 
+    async def consumed_record(self, key: MeterKey, jti: str) -> dict[str, Any] | None:
+        """The consumption record for the resource token's ``budget_consumed``
+        claim (draft §The Consumption Record): ``{"jti", "consumed"}`` for the
+        PRESENTED token — its total metered so far — or ``None`` when nothing
+        was metered against it. One record, the presented token's; the spend
+        under a person's other tokens is the usage endpoint's to report."""
+        async with self._lock:
+            pool = self._purge(key, time.monotonic())
+            if pool is None:
+                return None
+            total = pool.consumed.get(jti, 0)
+            return {"jti": jti, "consumed": total} if total > 0 else None
+
     async def consumed_records(self, key: MeterKey,
                                jkt: str | None = None) -> list[dict[str, Any]]:
-        """Per-token consumption for the resource token's ``budget_consumed``
-        claim: ``[{"jti": ..., "consumed": ...}, ...]``. Non-destructive — the
-        PS deduplicates by ``jti``, so reporting the same record twice is safe.
+        """Audit view: per-token consumption under this ledger key,
+        ``[{"jti": ..., "consumed": ...}, ...]``. Not what goes on the wire —
+        the resource token carries :meth:`consumed_record` — but the figures a
+        PS-side reconciliation or an operator wants to see.
 
         When ``jkt`` is given, records are scoped to tokens bound to that key:
         the agent carrying the resource token sees only its OWN spending, never

src/regent_httpsig/fastapi.py

@@ -122,7 +122,7 @@ async def require_signature(request: Request) -> VerifiedSignature:
 
 PriceFn = Callable[[Request], "int | None | Awaitable[int | None]"]
 ResourceTokenProvider = Callable[
-    [MeterKey, "list[dict[str, Any]]"], "str | None | Awaitable[str | None]"
+    [MeterKey, "dict[str, Any] | None"], "str | None | Awaitable[str | None]"
 ]
 
 
@@ -158,9 +158,12 @@ class BudgetMiddleware(BaseHTTPMiddleware):
       ``require=True`` refuses them with 401 + ``AAuth-Requirement``.
     - Error responses (4xx/5xx) release the reservation — nothing was served,
       the envelope is not charged.
-    - ``resource_token_provider(key, consumed_records)`` (optional) mints the
-      resource token embedded in budget-refusal responses so the agent can
-      carry ``budget_consumed`` back to its PS for re-authorization.
+    - ``resource_token_provider(key, record)`` (optional) mints the resource
+      token embedded in budget-refusal responses so the agent can carry
+      ``budget_consumed`` back to its PS for re-authorization. ``record`` is
+      the PRESENTED token's ``{"jti", "consumed"}`` (draft §The Consumption
+      Record — one record, two members) or ``None`` when nothing was metered
+      against it yet.
     """
 
     def __init__(
@@ -223,7 +226,7 @@ async def dispatch(
             reason = "budget-exhausted" if outcome.exhausted else "insufficient-budget"
             return await self._refusal_with_token(
                 reason=reason, envelope=envelope, remaining=outcome.remaining,
-                key=key, jkt=sig.keyid,
+                key=key, jti=jti,
                 # `required` rides only on insufficient-budget: what THIS
                 # request needed, so the agent can lower its bound and retry.
                 required=int(max_cost) if reason == "insufficient-budget" else None,
@@ -247,7 +250,7 @@ async def dispatch(
             # when `cost` is omitted — with `remaining` already net of the
             # hold, and commit when the stream completes. The agent recovers
             # the exact figure from the next response's `remaining`.
-            remaining = await self._meter.remaining(key)
+            remaining = await self._meter.remaining(key, jti)
             response.headers["AAuth-Budget"] = build_aauth_budget_header(
                 remaining=remaining, reserved=int(max_cost),
                 unit=envelope.unit, decimals=envelope.decimals,
@@ -311,14 +314,17 @@ async def _verified(self, request: Request) -> VerifiedSignature | None:
 
     async def _refusal_with_token(
         self, *, reason: str, envelope: BudgetClaim,
-        remaining: int, key: MeterKey, jkt: str | None = None,
+        remaining: int, key: MeterKey, jti: str,
         required: int | None = None,
     ) -> Response:
         token: str | None = None
         if self._resource_token is not None:
             try:
-                records = await self._meter.consumed_records(key, jkt=jkt)
-                token = await _maybe_await(self._resource_token(key, records))
+                # One record, the presented token's (§The Consumption Record):
+                # what THIS jti has cost so far. Siblings' spend never rides
+                # home with this agent — the usage endpoint reports that.
+                record = await self._meter.consumed_record(key, jti)
+                token = await _maybe_await(self._resource_token(key, record))
             except Exception:  # noqa: BLE001 — refusal must not fail on the extras
                 logger.warning("resource_token_provider failed", exc_info=True)
         return self._refusal(reason=reason, envelope=envelope,