feat(buyer-agent-registry): caching + rate-limit + audit emission (#380) (#407)

* feat(buyer-agent-registry): caching + rate-limit + audit emission (#380)

Three composable wrappers around the BuyerAgentRegistry Protocol that
the v3 seller's Tier 2 commercial-identity gate hits on every dispatch.
Without them, every resolve runs a SQL query — including negative paths
an enumeration probe will spam, making the lookup endpoint a
credential-stuffing oracle.

* CachingBuyerAgentRegistry — TTL + LRU cache, default 60s / 4096
  entries. Caches BOTH positive and negative resolutions; the negative
  cache closes the enumeration probe path so a probe walking arbitrary
  agent_url strings hits the DB at most once per (tenant, key) per TTL
  window. Hit-callback hook for Prometheus / OpenTelemetry counters.
* RateLimitedBuyerAgentRegistry — per-(tenant, lookup-key) token
  bucket, default 100 RPS. On exhaustion raises PERMISSION_DENIED with
  NO details and a generic message — wire-uniform with the registry-
  miss path from PR #393. A distinct RATE_LIMITED code or populated
  details would itself be an enumeration oracle.
* AuditingBuyerAgentRegistry — terminal wrapper emitting one
  AuditEvent per DB outcome (resolved / miss). The cache and rate-
  limit layers also accept the same audit_sink kwarg so cached_hit /
  cached_miss / rate_limited outcomes land in the same trail.

The wrappers stack outside-in. Adopters compose Caching(RateLimited(
Auditing(SQL-backed))) — the cache shortcuts repeated lookups before
the rate limiter or DB sees them; the rate limiter stops probe traffic
before the DB; the audit layer captures every actual lookup.

Wires the v3 reference seller's TenantScopedBuyerAgentRegistry into
the production stack with the same audit sink at every layer so SecOps
can reconstruct every resolve attempt. The make_registry factory
accepts ttl_seconds / rps_per_tenant / max_entries overrides for
adopters with different SLA / volume requirements.

Tests: tests/test_buyer_agent_registry_cache.py (23 tests covering
cache hit / miss / TTL expiry / LRU eviction / rate-limit threshold
+ refill / audit emission per outcome / sink-failure isolation /
end-to-end composition).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(buyer-agent-registry): hold lock in invalidate/clear (PR #407 fix-pack)

Code reviewer flagged that CachingBuyerAgentRegistry.invalidate() and
clear() mutated self._cache without holding self._lock. _store()'s
move_to_end / popitem(last=False) eviction races with concurrent
admin invalidate calls, risking RuntimeError or LRU-order corruption.

Convert both to async + acquire the lock before mutating. Test
updated to await the new coroutine.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* ci: retrigger

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bokelley

examples/v3_reference_seller/src/app.py

@@ -106,8 +106,13 @@ def main() -> None:
     asyncio.run(_bootstrap_schema(engine))
 
     router = SqlSubdomainTenantRouter(sessionmaker=sessionmaker)
-    buyer_registry = make_buyer_registry(sessionmaker)
     audit_sink = make_audit_sink(sessionmaker)
+    # The buyer registry composes cache + rate-limit + audit around
+    # the SQL-backed lookup. Wiring the same audit_sink at every
+    # layer means cached_hit / cached_miss / rate_limited / resolved
+    # / miss outcomes ALL land in the audit trail; SecOps can
+    # reconstruct every resolve attempt.
+    buyer_registry = make_buyer_registry(sessionmaker, audit_sink=audit_sink)
     # Anti-façade traffic recorder. The reference seller is a dev /
     # storyboard target, so we wire the in-memory recorder and flip
     # ``enable_debug_endpoints=True`` below to expose

examples/v3_reference_seller/src/buyer_registry.py

@@ -23,16 +23,21 @@
 
 from adcp.decisioning import (
     ApiKeyCredential,
+    AuditingBuyerAgentRegistry,
     BuyerAgent,
     BuyerAgentDefaultTerms,
     BuyerAgentRegistry,
+    CachingBuyerAgentRegistry,
     OAuthCredential,
+    RateLimitedBuyerAgentRegistry,
 )
 from adcp.server import current_tenant
 
 if TYPE_CHECKING:
     from sqlalchemy.ext.asyncio import async_sessionmaker
 
+    from adcp.audit_sink import AuditSink
+
 from .models import BuyerAgent as BuyerAgentRow
 
 logger = logging.getLogger(__name__)
@@ -130,11 +135,52 @@ def _row_to_agent(row: BuyerAgentRow) -> BuyerAgent:
     )
 
 
-def make_registry(sessionmaker: async_sessionmaker) -> BuyerAgentRegistry:
+def make_registry(
+    sessionmaker: async_sessionmaker,
+    *,
+    audit_sink: AuditSink | None = None,
+    ttl_seconds: float = 60.0,
+    rps_per_tenant: float = 100.0,
+    max_entries: int = 4096,
+) -> BuyerAgentRegistry:
     """Factory for the tenant-scoped registry. Returns a
     Protocol-typed handle — adopters wire it into
-    :func:`adcp.decisioning.serve` via ``buyer_agent_registry=``."""
-    return TenantScopedBuyerAgentRegistry(sessionmaker=sessionmaker)
+    :func:`adcp.decisioning.serve` via ``buyer_agent_registry=``.
+
+    Composes three wrappers around the SQL-backed registry so the
+    Tier 2 commercial-identity gate has production-grade properties:
+
+    * **Cache** (outermost) — TTL + LRU. Both positive and negative
+      resolutions cached so an enumeration probe at the lookup
+      endpoint hits the DB at most once per ``(tenant, key)`` per
+      ``ttl_seconds`` window.
+    * **Rate limit** (middle) — token bucket per
+      ``(tenant, lookup_key)``. On exhaustion raises
+      ``PERMISSION_DENIED`` with no ``details`` so the wire shape
+      matches the registry-miss path (no enumeration oracle).
+    * **Audit** (innermost) — emits one
+      :class:`~adcp.audit_sink.AuditEvent` per DB outcome
+      (``resolved`` / ``miss``); the cache and rate-limit layers
+      add ``cached_hit`` / ``cached_miss`` / ``rate_limited``
+      events to the same sink so compliance teams reconstruct
+      every resolve attempt.
+
+    Adopters needing different defaults pass ``ttl_seconds`` /
+    ``rps_per_tenant`` / ``max_entries`` overrides.
+    """
+    sql_backed = TenantScopedBuyerAgentRegistry(sessionmaker=sessionmaker)
+    audited = AuditingBuyerAgentRegistry(sql_backed, audit_sink=audit_sink)
+    rate_limited = RateLimitedBuyerAgentRegistry(
+        audited,
+        rps_per_tenant=rps_per_tenant,
+        audit_sink=audit_sink,
+    )
+    return CachingBuyerAgentRegistry(
+        rate_limited,
+        ttl_seconds=ttl_seconds,
+        max_entries=max_entries,
+        audit_sink=audit_sink,
+    )
 
 
 __all__ = ["TenantScopedBuyerAgentRegistry", "make_registry"]

src/adcp/decisioning/__init__.py

@@ -87,6 +87,11 @@ def create_media_buy(
     signing_only_registry,
     validate_billing_for_agent,
 )
+from adcp.decisioning.registry_cache import (
+    AuditingBuyerAgentRegistry,
+    CachingBuyerAgentRegistry,
+    RateLimitedBuyerAgentRegistry,
+)
 from adcp.decisioning.resolve import (
     CollectionList,
     Format,
@@ -173,13 +178,15 @@ def __init__(self, *args: object, **kwargs: object) -> None:
     "AdcpError",
     "ApiKeyCredential",
     "AudiencePlatform",
+    "AuditingBuyerAgentRegistry",
     "AuthInfo",
     "BillingMode",
     "BrandRightsPlatform",
     "BuyerAgent",
     "BuyerAgentDefaultTerms",
     "BuyerAgentRegistry",
     "BuyerAgentStatus",
+    "CachingBuyerAgentRegistry",
     "CampaignGovernancePlatform",
     "CollectionList",
     "CollectionListsPlatform",
@@ -207,6 +214,7 @@ def __init__(self, *args: object, **kwargs: object) -> None:
     "PropertyList",
     "PropertyListReference",
     "PropertyListsPlatform",
+    "RateLimitedBuyerAgentRegistry",
     "RequestContext",
     "ResourceResolver",
     "SalesPlatform",