Module adcp.decisioning.registry_cache
In-process wrappers around :class:~adcp.decisioning.BuyerAgentRegistry.
The Tier 2 commercial-identity gate fires on every dispatched skill.
A bare :class:PgBuyerAgentRegistry (or the v3 reference seller's
:class:TenantScopedBuyerAgentRegistry) hits the database on each
call — including the negative paths an enumeration probe will spam.
This module supplies three composable wrappers that implement the
:class:BuyerAgentRegistry Protocol and stack arbitrarily:
- :class:
CachingBuyerAgentRegistry— TTL + LRU cache for positive AND negative resolutions. Negative caching is the load-shaping move: an enumeration probe walking a millionagent_urlstrings would otherwise hit the DB once per probe; with negative caching it hits the DB once per(tenant, agent_url)pair within the TTL window. - :class:
RateLimitedBuyerAgentRegistry— per-(tenant, lookup-key) token bucket. On exhaustion, raisesPERMISSION_DENIEDwith nodetailsso the wire shape matches every other denied path (registry miss, suspended, blocked) — preserves the spec's omit-on-unestablished-identity rule from PR #393. A distinctRATE_LIMITEDcode would itself be an enumeration oracle. - :class:
AuditingBuyerAgentRegistry— terminal wrapper that fires one :class:~adcp.audit_sink.AuditEventperresolved/missoutcome from the inner store. Combine with the per-layeraudit_sinkkwarg on the cache / rate-limit wrappers to capturecached_hit/cached_miss/rate_limitedevents too.
Every wrapper accepts an optional audit_sink. When provided, the
wrapper emits one :class:AuditEvent for any outcome it terminates
the call on (cache hit, rate-limit reject, DB resolve). Outcomes that
fall through to the inner wrapper are NOT re-emitted by the outer —
ordering avoids double-counting. If no sink is wired the outcome
logs at DEBUG.
Composition
The wrappers stack outside-in. Adopters typically build::
inner = AuditingBuyerAgentRegistry(
sql_backed_registry, # actual DB lookup
audit_sink=sink,
)
rate_limited = RateLimitedBuyerAgentRegistry(
inner,
rps_per_tenant=100,
audit_sink=sink, # capture rate_limited events
)
registry = CachingBuyerAgentRegistry(
rate_limited,
ttl_seconds=60,
audit_sink=sink, # capture cached_hit/cached_miss events
)
Order matters:
- Cache is OUTERMOST so cached hits skip rate-limit accounting and the DB. Negative cache prevents the credential-stuffing oracle on the lookup endpoint.
- Rate limit sits BETWEEN cache and inner so the limiter only fires on calls that actually need DB work. Cached hits don't burn tokens.
- The terminal :class:
AuditingBuyerAgentRegistryrecords the actual DB outcome.
Tenant Scoping
The cache and rate limiter both key on (tenant_id, lookup_key).
The tenant id comes from :func:current_tenant() — set
by the adopter's tenant middleware before the framework dispatches.
Adopters running single-tenant skip this contextvar; tenant_id
falls through as None and the wrappers behave as a flat keyspace.
Global variables
var MetricCallback-
Type alias for the metric callback. Counters are incremented by 1 each call. Adopters wire this to Prometheus, OpenTelemetry, StatsD, etc.
Classes
class AuditingBuyerAgentRegistry (inner: BuyerAgentRegistry,
*,
audit_sink: AuditSink | None = None,
sink_timeout_seconds: float = 5.0)-
Expand source code
class AuditingBuyerAgentRegistry: """Terminal wrapper that emits one :class:`AuditEvent` per resolution outcome from the inner store. Wrap the SQL-backed registry with this so every DB lookup (``resolved`` / ``miss``) lands in the audit trail. Compliance teams reconstruct who tried what when from these records; SecOps correlates spikes in ``miss`` events with credential-stuffing activity. The event ``operation`` is namespaced (``"buyer_agent_registry.resolve_by_agent_url"`` / ``"...resolve_by_credential"``) so audit queries can filter registry traffic from the higher-level skill dispatches. :param inner: Wrapped :class:`BuyerAgentRegistry` — typically the actual SQL-backed impl. :param audit_sink: :class:`AuditSink` to write events to. If ``None``, outcomes log at ``DEBUG`` instead. :param sink_timeout_seconds: Per-sink timeout. Default 5s matching :func:`adcp.audit_sink.make_audit_middleware`. A sink that wedges (DB stall, S3 outage) NEVER blocks dispatch. """ def __init__( self, inner: BuyerAgentRegistry, *, audit_sink: AuditSink | None = None, sink_timeout_seconds: float = 5.0, ) -> None: self._inner = inner self._sink = audit_sink self._sink_timeout = sink_timeout_seconds async def resolve_by_agent_url(self, agent_url: str) -> BuyerAgent | None: tenant_id = _current_tenant_id() result = await self._inner.resolve_by_agent_url(agent_url) await _emit_audit( self._sink, operation="buyer_agent_registry.resolve_by_agent_url", outcome="resolved" if result is not None else "miss", lookup_key=f"agent_url:{agent_url}", tenant_id=tenant_id, agent=result, sink_timeout_seconds=self._sink_timeout, ) return result async def resolve_by_credential(self, credential: Credential) -> BuyerAgent | None: tenant_id = _current_tenant_id() result = await self._inner.resolve_by_credential(credential) await _emit_audit( self._sink, operation="buyer_agent_registry.resolve_by_credential", outcome="resolved" if result is not None else "miss", lookup_key=_credential_key(credential), tenant_id=tenant_id, agent=result, sink_timeout_seconds=self._sink_timeout, ) return resultTerminal wrapper that emits one :class:
AuditEventper resolution outcome from the inner store.Wrap the SQL-backed registry with this so every DB lookup (
resolved/miss) lands in the audit trail. Compliance teams reconstruct who tried what when from these records; SecOps correlates spikes inmissevents with credential-stuffing activity.The event
operationis namespaced ("buyer_agent_registry.resolve_by_agent_url"/"...resolve_by_credential") so audit queries can filter registry traffic from the higher-level skill dispatches.:param inner: Wrapped :class:
BuyerAgentRegistry— typically the actual SQL-backed impl. :param audit_sink: :class:AuditSinkto write events to. IfNone, outcomes log atDEBUGinstead. :param sink_timeout_seconds: Per-sink timeout. Default 5s matching :func:make_audit_middleware(). A sink that wedges (DB stall, S3 outage) NEVER blocks dispatch.Methods
async def resolve_by_agent_url(self,
agent_url: str) ‑> BuyerAgent | None-
Expand source code
async def resolve_by_agent_url(self, agent_url: str) -> BuyerAgent | None: tenant_id = _current_tenant_id() result = await self._inner.resolve_by_agent_url(agent_url) await _emit_audit( self._sink, operation="buyer_agent_registry.resolve_by_agent_url", outcome="resolved" if result is not None else "miss", lookup_key=f"agent_url:{agent_url}", tenant_id=tenant_id, agent=result, sink_timeout_seconds=self._sink_timeout, ) return result async def resolve_by_credential(self, credential: Credential) ‑> BuyerAgent | None-
Expand source code
async def resolve_by_credential(self, credential: Credential) -> BuyerAgent | None: tenant_id = _current_tenant_id() result = await self._inner.resolve_by_credential(credential) await _emit_audit( self._sink, operation="buyer_agent_registry.resolve_by_credential", outcome="resolved" if result is not None else "miss", lookup_key=_credential_key(credential), tenant_id=tenant_id, agent=result, sink_timeout_seconds=self._sink_timeout, ) return result
class CachingBuyerAgentRegistry (inner: BuyerAgentRegistry,
*,
ttl_seconds: float = 60.0,
max_entries: int = 4096,
hit_callback: MetricCallback | None = None,
audit_sink: AuditSink | None = None,
sink_timeout_seconds: float = 5.0,
time_source: Callable[[], float] = <built-in function monotonic>)-
Expand source code
class CachingBuyerAgentRegistry: """In-process TTL + LRU cache wrapping any :class:`BuyerAgentRegistry`. Caches BOTH positive and negative resolutions. Negative caching is the load-shaping move — an enumeration probe walking arbitrary ``agent_url`` strings would otherwise hit the DB once per probe; with negative caching it hits the DB at most once per ``(tenant, agent_url)`` pair within the TTL window. :param inner: The wrapped :class:`BuyerAgentRegistry` — typically a :class:`RateLimitedBuyerAgentRegistry` or a SQL-backed impl. :param ttl_seconds: How long a resolution stays cached. Default 60s — long enough to absorb burst traffic during a media-buy flight, short enough that a status flip (active → suspended) propagates within a minute. :param max_entries: LRU cap. Default 4096 — bounded so an enumeration probe can't blow up memory, large enough that steady-state hot agents stay resident. :param hit_callback: Optional counter-style metric hook fired on every cache event. Receives ``"hit"`` / ``"miss"`` / ``"negative_hit"`` / ``"expired"``. :param audit_sink: Optional audit sink — emits ``cached_hit`` / ``cached_miss`` events for served-from-cache outcomes. Misses fall through to the inner wrapper which emits its own event for the actual resolution. :param time_source: Override for tests — defaults to :func:`time.monotonic`. Use a fake clock to drive TTL expiry. Concurrency ----------- The cache uses an ``asyncio.Lock`` to serialize entry insertion and LRU promotion. The lock is held only across the dict update — the inner ``resolve_*`` call happens OUTSIDE the lock, so concurrent misses against different keys race the inner backend in parallel. A burst of concurrent misses against the SAME key will all hit the backend (no thundering-herd dedup) — adopters needing single-flight wrap with their own dedup. The simpler "all races to the backend" shape avoids the complexity of an in-flight registry while accepting bounded duplicate work. """ def __init__( self, inner: BuyerAgentRegistry, *, ttl_seconds: float = 60.0, max_entries: int = 4096, hit_callback: MetricCallback | None = None, audit_sink: AuditSink | None = None, sink_timeout_seconds: float = 5.0, time_source: Callable[[], float] = time.monotonic, ) -> None: if ttl_seconds <= 0: raise ValueError(f"ttl_seconds must be > 0, got {ttl_seconds!r}") if max_entries <= 0: raise ValueError(f"max_entries must be > 0, got {max_entries!r}") self._inner = inner self._ttl = ttl_seconds self._max = max_entries self._hit_cb = hit_callback self._sink = audit_sink self._sink_timeout = sink_timeout_seconds self._now = time_source # OrderedDict gives us O(1) move-to-end for LRU semantics on # every hit, plus O(1) popitem(last=False) for eviction. self._cache: OrderedDict[tuple[str | None, str], _CacheEntry] = OrderedDict() self._lock = asyncio.Lock() async def resolve_by_agent_url(self, agent_url: str) -> BuyerAgent | None: """Resolve via cache, falling through to ``inner`` on miss.""" tenant_id = _current_tenant_id() lookup_key = f"agent_url:{agent_url}" key = (tenant_id, lookup_key) cached = await self._lookup(key) if cached is not None: await _emit_audit( self._sink, operation="buyer_agent_registry.resolve_by_agent_url", outcome="cached_hit" if cached.value is not None else "cached_miss", lookup_key=lookup_key, tenant_id=tenant_id, agent=cached.value, sink_timeout_seconds=self._sink_timeout, ) return cached.value result = await self._inner.resolve_by_agent_url(agent_url) await self._store(key, result) return result async def resolve_by_credential(self, credential: Credential) -> BuyerAgent | None: """Resolve via cache, falling through to ``inner`` on miss.""" tenant_id = _current_tenant_id() lookup_key = _credential_key(credential) key = (tenant_id, lookup_key) cached = await self._lookup(key) if cached is not None: await _emit_audit( self._sink, operation="buyer_agent_registry.resolve_by_credential", outcome="cached_hit" if cached.value is not None else "cached_miss", lookup_key=lookup_key, tenant_id=tenant_id, agent=cached.value, sink_timeout_seconds=self._sink_timeout, ) return cached.value result = await self._inner.resolve_by_credential(credential) await self._store(key, result) return result async def _lookup(self, key: tuple[str | None, str]) -> _CacheEntry | None: """Return a non-expired cache entry for ``key`` or ``None``. Lock held only for the dict mutation; the entry itself is immutable so dropping the lock before returning is safe. """ async with self._lock: entry = self._cache.get(key) if entry is None: self._fire("miss") return None if entry.expires_at <= self._now(): # Expired — evict and treat as miss so the caller # re-fetches from the inner backend. del self._cache[key] self._fire("expired") return None # Promote to most-recently-used. self._cache.move_to_end(key) self._fire("hit" if entry.value is not None else "negative_hit") return entry async def _store(self, key: tuple[str | None, str], value: BuyerAgent | None) -> None: async with self._lock: self._cache[key] = _CacheEntry( value=value, expires_at=self._now() + self._ttl, ) self._cache.move_to_end(key) while len(self._cache) > self._max: self._cache.popitem(last=False) def _fire(self, label: str) -> None: if self._hit_cb is not None: try: self._hit_cb(label) except Exception: # noqa: BLE001 — metric callback must not break dispatch logger.warning( "[adcp.registry_cache] hit_callback raised for label=%s", label, exc_info=True, ) async def invalidate(self, *, tenant_id: str | None, lookup_key: str) -> None: """Drop a single ``(tenant_id, lookup_key)`` entry. Called by admin / management code on a status flip — e.g., when an operator suspends an agent the management API calls ``invalidate`` so the next dispatch sees the new status immediately rather than waiting for TTL expiry. Async + lock-held because admin paths run concurrently with dispatch traffic; mutating the underlying ``OrderedDict`` while ``_store`` is reordering or evicting can corrupt the LRU order or raise ``RuntimeError: OrderedDict mutated during iteration``. """ async with self._lock: self._cache.pop((tenant_id, lookup_key), None) def clear_sync(self) -> None: """Drop every cached entry from a sync context. Safe to call from any thread or coroutine without an event loop. Atomic via the GIL on :meth:`OrderedDict.clear` — no lock acquired, so a concurrent async ``_lookup`` / ``_store`` may observe either the pre-clear or post-clear dict. The worst case is one extra round-trip to the inner registry on the next resolve, which is exactly what an invalidation is supposed to cause. Use cases: mutation-observer hooks wired by :meth:`PgBuyerAgentRegistry.with_caching`, post-config-reload flushes from sync admin code. Full-clear (rather than per-key drop) trades a small amount of over-invalidation for simplicity: mutations are admin-rare and the next traffic burst rebuilds the working set within TTL. Adopters needing finer-grained invalidation still have :meth:`invalidate` for explicit ``(tenant_id, lookup_key)`` drops. """ self._cache.clear() async def clear(self) -> None: """Drop every cached entry. For tests + post-config-reload. Async + lock-held for the same reason as :meth:`invalidate`. """ async with self._lock: self._cache.clear()In-process TTL + LRU cache wrapping any :class:
BuyerAgentRegistry.Caches BOTH positive and negative resolutions. Negative caching is the load-shaping move — an enumeration probe walking arbitrary
agent_urlstrings would otherwise hit the DB once per probe; with negative caching it hits the DB at most once per(tenant, agent_url)pair within the TTL window.:param inner: The wrapped :class:
BuyerAgentRegistry— typically a :class:RateLimitedBuyerAgentRegistryor a SQL-backed impl. :param ttl_seconds: How long a resolution stays cached. Default 60s — long enough to absorb burst traffic during a media-buy flight, short enough that a status flip (active → suspended) propagates within a minute. :param max_entries: LRU cap. Default 4096 — bounded so an enumeration probe can't blow up memory, large enough that steady-state hot agents stay resident. :param hit_callback: Optional counter-style metric hook fired on every cache event. Receives"hit"/"miss"/"negative_hit"/"expired". :param audit_sink: Optional audit sink — emitscached_hit/cached_missevents for served-from-cache outcomes. Misses fall through to the inner wrapper which emits its own event for the actual resolution. :param time_source: Override for tests — defaults to :func:time.monotonic. Use a fake clock to drive TTL expiry.Concurrency
The cache uses an
asyncio.Lockto serialize entry insertion and LRU promotion. The lock is held only across the dict update — the innerresolve_*call happens OUTSIDE the lock, so concurrent misses against different keys race the inner backend in parallel. A burst of concurrent misses against the SAME key will all hit the backend (no thundering-herd dedup) — adopters needing single-flight wrap with their own dedup. The simpler "all races to the backend" shape avoids the complexity of an in-flight registry while accepting bounded duplicate work.Methods
async def clear(self) ‑> None-
Expand source code
async def clear(self) -> None: """Drop every cached entry. For tests + post-config-reload. Async + lock-held for the same reason as :meth:`invalidate`. """ async with self._lock: self._cache.clear()Drop every cached entry. For tests + post-config-reload.
Async + lock-held for the same reason as :meth:
invalidate. def clear_sync(self) ‑> None-
Expand source code
def clear_sync(self) -> None: """Drop every cached entry from a sync context. Safe to call from any thread or coroutine without an event loop. Atomic via the GIL on :meth:`OrderedDict.clear` — no lock acquired, so a concurrent async ``_lookup`` / ``_store`` may observe either the pre-clear or post-clear dict. The worst case is one extra round-trip to the inner registry on the next resolve, which is exactly what an invalidation is supposed to cause. Use cases: mutation-observer hooks wired by :meth:`PgBuyerAgentRegistry.with_caching`, post-config-reload flushes from sync admin code. Full-clear (rather than per-key drop) trades a small amount of over-invalidation for simplicity: mutations are admin-rare and the next traffic burst rebuilds the working set within TTL. Adopters needing finer-grained invalidation still have :meth:`invalidate` for explicit ``(tenant_id, lookup_key)`` drops. """ self._cache.clear()Drop every cached entry from a sync context.
Safe to call from any thread or coroutine without an event loop. Atomic via the GIL on :meth:
OrderedDict.clear— no lock acquired, so a concurrent async_lookup/_storemay observe either the pre-clear or post-clear dict. The worst case is one extra round-trip to the inner registry on the next resolve, which is exactly what an invalidation is supposed to cause.Use cases: mutation-observer hooks wired by :meth:
PgBuyerAgentRegistry.with_caching, post-config-reload flushes from sync admin code.Full-clear (rather than per-key drop) trades a small amount of over-invalidation for simplicity: mutations are admin-rare and the next traffic burst rebuilds the working set within TTL. Adopters needing finer-grained invalidation still have :meth:
invalidatefor explicit(tenant_id, lookup_key)drops. async def invalidate(self,
*,
tenant_id: str | None,
lookup_key: str) ‑> None-
Expand source code
async def invalidate(self, *, tenant_id: str | None, lookup_key: str) -> None: """Drop a single ``(tenant_id, lookup_key)`` entry. Called by admin / management code on a status flip — e.g., when an operator suspends an agent the management API calls ``invalidate`` so the next dispatch sees the new status immediately rather than waiting for TTL expiry. Async + lock-held because admin paths run concurrently with dispatch traffic; mutating the underlying ``OrderedDict`` while ``_store`` is reordering or evicting can corrupt the LRU order or raise ``RuntimeError: OrderedDict mutated during iteration``. """ async with self._lock: self._cache.pop((tenant_id, lookup_key), None)Drop a single
(tenant_id, lookup_key)entry.Called by admin / management code on a status flip — e.g., when an operator suspends an agent the management API calls
invalidateso the next dispatch sees the new status immediately rather than waiting for TTL expiry.Async + lock-held because admin paths run concurrently with dispatch traffic; mutating the underlying
OrderedDictwhile_storeis reordering or evicting can corrupt the LRU order or raiseRuntimeError: OrderedDict mutated during iteration. async def resolve_by_agent_url(self,
agent_url: str) ‑> BuyerAgent | None-
Expand source code
async def resolve_by_agent_url(self, agent_url: str) -> BuyerAgent | None: """Resolve via cache, falling through to ``inner`` on miss.""" tenant_id = _current_tenant_id() lookup_key = f"agent_url:{agent_url}" key = (tenant_id, lookup_key) cached = await self._lookup(key) if cached is not None: await _emit_audit( self._sink, operation="buyer_agent_registry.resolve_by_agent_url", outcome="cached_hit" if cached.value is not None else "cached_miss", lookup_key=lookup_key, tenant_id=tenant_id, agent=cached.value, sink_timeout_seconds=self._sink_timeout, ) return cached.value result = await self._inner.resolve_by_agent_url(agent_url) await self._store(key, result) return resultResolve via cache, falling through to
inneron miss. async def resolve_by_credential(self, credential: Credential) ‑> BuyerAgent | None-
Expand source code
async def resolve_by_credential(self, credential: Credential) -> BuyerAgent | None: """Resolve via cache, falling through to ``inner`` on miss.""" tenant_id = _current_tenant_id() lookup_key = _credential_key(credential) key = (tenant_id, lookup_key) cached = await self._lookup(key) if cached is not None: await _emit_audit( self._sink, operation="buyer_agent_registry.resolve_by_credential", outcome="cached_hit" if cached.value is not None else "cached_miss", lookup_key=lookup_key, tenant_id=tenant_id, agent=cached.value, sink_timeout_seconds=self._sink_timeout, ) return cached.value result = await self._inner.resolve_by_credential(credential) await self._store(key, result) return resultResolve via cache, falling through to
inneron miss.
class RateLimitedBuyerAgentRegistry (inner: BuyerAgentRegistry,
*,
rps_per_tenant: float = 100.0,
burst: float | None = None,
audit_sink: AuditSink | None = None,
sink_timeout_seconds: float = 5.0,
time_source: Callable[[], float] = <built-in function monotonic>)-
Expand source code
class RateLimitedBuyerAgentRegistry: """Per-tenant token-bucket rate limiter wrapping a :class:`BuyerAgentRegistry`. Sized for the credential-stuffing oracle: the registry's :meth:`resolve_by_credential` is queryable with arbitrary ``key_id`` strings; without a rate limit, an attacker can enumerate the keyspace at line rate. The bucket sits between the request and the DB so probe traffic gets rejected before the SQL query runs. :param inner: The wrapped :class:`BuyerAgentRegistry`. :param rps_per_tenant: Steady-state requests per second per ``(tenant_id, lookup_key)`` bucket. Default 100 — high enough to absorb a real buyer's storyboard burst, low enough that an enumeration probe at line rate gets cut off. :param burst: Maximum bucket capacity (tokens). Default ``rps_per_tenant`` so a steady state can sustain ``rps_per_tenant`` calls/sec but bursts are capped at the same number. Adopters with bursty real traffic raise this. :param audit_sink: Optional audit sink — emits ``rate_limited`` events when the bucket is exhausted. The most interesting event for security review (repeated rate-limit exhaustion is the credential-stuffing signal an attacker is actively probing). :param time_source: Override for tests — defaults to :func:`time.monotonic`. Failure mode ------------ On bucket exhaustion, raises :class:`AdcpError` ``PERMISSION_DENIED`` with NO ``details`` and a generic message matching the registry-miss path. This is deliberate — a distinct ``RATE_LIMITED`` code or any populated ``details`` field would itself be an enumeration oracle (the attacker learns "this ``agent_url`` is interesting enough to be rate-limited"). The spec's omit-on-unestablished-identity rule from PR #393 applies. """ def __init__( self, inner: BuyerAgentRegistry, *, rps_per_tenant: float = 100.0, burst: float | None = None, audit_sink: AuditSink | None = None, sink_timeout_seconds: float = 5.0, time_source: Callable[[], float] = time.monotonic, ) -> None: if rps_per_tenant <= 0: raise ValueError(f"rps_per_tenant must be > 0, got {rps_per_tenant!r}") if burst is not None and burst <= 0: raise ValueError(f"burst must be > 0, got {burst!r}") self._inner = inner self._rate = rps_per_tenant self._burst = burst if burst is not None else rps_per_tenant self._sink = audit_sink self._sink_timeout = sink_timeout_seconds self._now = time_source self._buckets: dict[tuple[str | None, str], _Bucket] = {} self._lock = asyncio.Lock() async def resolve_by_agent_url(self, agent_url: str) -> BuyerAgent | None: tenant_id = _current_tenant_id() lookup_key = f"agent_url:{agent_url}" await self._charge( (tenant_id, lookup_key), operation="buyer_agent_registry.resolve_by_agent_url", tenant_id=tenant_id, ) return await self._inner.resolve_by_agent_url(agent_url) async def resolve_by_credential(self, credential: Credential) -> BuyerAgent | None: tenant_id = _current_tenant_id() lookup_key = _credential_key(credential) await self._charge( (tenant_id, lookup_key), operation="buyer_agent_registry.resolve_by_credential", tenant_id=tenant_id, ) return await self._inner.resolve_by_credential(credential) async def _charge( self, key: tuple[str | None, str], *, operation: str, tenant_id: str | None, ) -> None: """Spend one token from ``key``'s bucket; raise + audit on exhaustion.""" now = self._now() async with self._lock: bucket = self._buckets.get(key) if bucket is None: # New bucket — start full so a fresh tenant gets the # burst allowance immediately. bucket = _Bucket(tokens=self._burst, last_refill=now) self._buckets[key] = bucket else: # Refill at ``rate`` tokens/sec, capped at ``burst``. elapsed = now - bucket.last_refill bucket.tokens = min(self._burst, bucket.tokens + elapsed * self._rate) bucket.last_refill = now if bucket.tokens < 1.0: exhausted = True else: bucket.tokens -= 1.0 exhausted = False if exhausted: # Audit emission OUTSIDE the lock — the sink may be slow. await _emit_audit( self._sink, operation=operation, outcome="rate_limited", lookup_key=key[1], tenant_id=tenant_id, sink_timeout_seconds=self._sink_timeout, ) raise _denied_error()Per-tenant token-bucket rate limiter wrapping a :class:
BuyerAgentRegistry.Sized for the credential-stuffing oracle: the registry's :meth:
resolve_by_credentialis queryable with arbitrarykey_idstrings; without a rate limit, an attacker can enumerate the keyspace at line rate. The bucket sits between the request and the DB so probe traffic gets rejected before the SQL query runs.:param inner: The wrapped :class:
BuyerAgentRegistry. :param rps_per_tenant: Steady-state requests per second per(tenant_id, lookup_key)bucket. Default 100 — high enough to absorb a real buyer's storyboard burst, low enough that an enumeration probe at line rate gets cut off. :param burst: Maximum bucket capacity (tokens). Defaultrps_per_tenantso a steady state can sustainrps_per_tenantcalls/sec but bursts are capped at the same number. Adopters with bursty real traffic raise this. :param audit_sink: Optional audit sink — emitsrate_limitedevents when the bucket is exhausted. The most interesting event for security review (repeated rate-limit exhaustion is the credential-stuffing signal an attacker is actively probing). :param time_source: Override for tests — defaults to :func:time.monotonic.Failure Mode
On bucket exhaustion, raises :class:
AdcpErrorPERMISSION_DENIEDwith NOdetailsand a generic message matching the registry-miss path. This is deliberate — a distinctRATE_LIMITEDcode or any populateddetailsfield would itself be an enumeration oracle (the attacker learns "thisagent_urlis interesting enough to be rate-limited"). The spec's omit-on-unestablished-identity rule from PR #393 applies.Methods
async def resolve_by_agent_url(self,
agent_url: str) ‑> BuyerAgent | None-
Expand source code
async def resolve_by_agent_url(self, agent_url: str) -> BuyerAgent | None: tenant_id = _current_tenant_id() lookup_key = f"agent_url:{agent_url}" await self._charge( (tenant_id, lookup_key), operation="buyer_agent_registry.resolve_by_agent_url", tenant_id=tenant_id, ) return await self._inner.resolve_by_agent_url(agent_url) async def resolve_by_credential(self, credential: Credential) ‑> BuyerAgent | None-
Expand source code
async def resolve_by_credential(self, credential: Credential) -> BuyerAgent | None: tenant_id = _current_tenant_id() lookup_key = _credential_key(credential) await self._charge( (tenant_id, lookup_key), operation="buyer_agent_registry.resolve_by_credential", tenant_id=tenant_id, ) return await self._inner.resolve_by_credential(credential)
class ResolveOutcome (...)-
str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.
Subclasses
- a2a.compat.v0_3.types.In
- a2a.compat.v0_3.types.Role
- a2a.compat.v0_3.types.TaskState
- a2a.compat.v0_3.types.TransportProtocol
- a2a.utils.constants.TransportProtocol
- CanonicalReferenceStatus
- ProposalState
- ActivityType
- Protocol
- TaskStatus
- UnknownFieldPolicy
- enum.StrEnum
- importlib.metadata._text.FoldedCase
- markdown.util.AtomicString
- markupsafe.Markup
- pydantic._internal._repr.PlainRepr
- pydantic.types.PaymentCardBrand
- pydantic.types.PaymentCardNumber
- pydantic_settings.sources.types.EnvNoneType
- starlette.datastructures.URLPath
Static methods
def maketrans(...)-
Return a translation table usable for str.translate().
If there is only one argument, it must be a dictionary mapping Unicode ordinals (integers) or characters to Unicode ordinals, strings or None. Character keys will be then converted to ordinals. If there are two arguments, they must be strings of equal length, and in the resulting dictionary, each character in x will be mapped to the character at the same position in y. If there is a third argument, it must be a string, whose characters will be mapped to None in the result.
Methods
def capitalize(self, /)-
Return a capitalized version of the string.
More specifically, make the first character have upper case and the rest lower case.
def casefold(self, /)-
Return a version of the string suitable for caseless comparisons.
def center(self, width, fillchar=' ', /)-
Return a centered string of length width.
Padding is done using the specified fill character (default is a space).
def count(...)-
S.count(sub[, start[, end]]) -> int
Return the number of non-overlapping occurrences of substring sub in string S[start:end]. Optional arguments start and end are interpreted as in slice notation.
def encode(self, /, encoding='utf-8', errors='strict')-
Encode the string using the codec registered for encoding.
encoding The encoding in which to encode the string. errors The error handling scheme to use for encoding errors. The default is 'strict' meaning that encoding errors raise a UnicodeEncodeError. Other possible values are 'ignore', 'replace' and 'xmlcharrefreplace' as well as any other name registered with codecs.register_error that can handle UnicodeEncodeErrors.
def endswith(...)-
S.endswith(suffix[, start[, end]]) -> bool
Return True if S ends with the specified suffix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. suffix can also be a tuple of strings to try.
def expandtabs(self, /, tabsize=8)-
Return a copy where all tab characters are expanded using spaces.
If tabsize is not given, a tab size of 8 characters is assumed.
def find(...)-
S.find(sub[, start[, end]]) -> int
Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.
Return -1 on failure.
def format(...)-
S.format(args, *kwargs) -> str
Return a formatted version of S, using substitutions from args and kwargs. The substitutions are identified by braces ('{' and '}').
def format_map(...)-
S.format_map(mapping) -> str
Return a formatted version of S, using substitutions from mapping. The substitutions are identified by braces ('{' and '}').
def index(...)-
S.index(sub[, start[, end]]) -> int
Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.
Raises ValueError when the substring is not found.
def isalnum(self, /)-
Return True if the string is an alpha-numeric string, False otherwise.
A string is alpha-numeric if all characters in the string are alpha-numeric and there is at least one character in the string.
def isalpha(self, /)-
Return True if the string is an alphabetic string, False otherwise.
A string is alphabetic if all characters in the string are alphabetic and there is at least one character in the string.
def isascii(self, /)-
Return True if all characters in the string are ASCII, False otherwise.
ASCII characters have code points in the range U+0000-U+007F. Empty string is ASCII too.
def isdecimal(self, /)-
Return True if the string is a decimal string, False otherwise.
A string is a decimal string if all characters in the string are decimal and there is at least one character in the string.
def isdigit(self, /)-
Return True if the string is a digit string, False otherwise.
A string is a digit string if all characters in the string are digits and there is at least one character in the string.
def isidentifier(self, /)-
Return True if the string is a valid Python identifier, False otherwise.
Call keyword.iskeyword(s) to test whether string s is a reserved identifier, such as "def" or "class".
def islower(self, /)-
Return True if the string is a lowercase string, False otherwise.
A string is lowercase if all cased characters in the string are lowercase and there is at least one cased character in the string.
def isnumeric(self, /)-
Return True if the string is a numeric string, False otherwise.
A string is numeric if all characters in the string are numeric and there is at least one character in the string.
def isprintable(self, /)-
Return True if the string is printable, False otherwise.
A string is printable if all of its characters are considered printable in repr() or if it is empty.
def isspace(self, /)-
Return True if the string is a whitespace string, False otherwise.
A string is whitespace if all characters in the string are whitespace and there is at least one character in the string.
def istitle(self, /)-
Return True if the string is a title-cased string, False otherwise.
In a title-cased string, upper- and title-case characters may only follow uncased characters and lowercase characters only cased ones.
def isupper(self, /)-
Return True if the string is an uppercase string, False otherwise.
A string is uppercase if all cased characters in the string are uppercase and there is at least one cased character in the string.
def join(self, iterable, /)-
Concatenate any number of strings.
The string whose method is called is inserted in between each given string. The result is returned as a new string.
Example: '.'.join(['ab', 'pq', 'rs']) -> 'ab.pq.rs'
def ljust(self, width, fillchar=' ', /)-
Return a left-justified string of length width.
Padding is done using the specified fill character (default is a space).
def lower(self, /)-
Return a copy of the string converted to lowercase.
def lstrip(self, chars=None, /)-
Return a copy of the string with leading whitespace removed.
If chars is given and not None, remove characters in chars instead.
def partition(self, sep, /)-
Partition the string into three parts using the given separator.
This will search for the separator in the string. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.
If the separator is not found, returns a 3-tuple containing the original string and two empty strings.
def removeprefix(self, prefix, /)-
Return a str with the given prefix string removed if present.
If the string starts with the prefix string, return string[len(prefix):]. Otherwise, return a copy of the original string.
def removesuffix(self, suffix, /)-
Return a str with the given suffix string removed if present.
If the string ends with the suffix string and that suffix is not empty, return string[:-len(suffix)]. Otherwise, return a copy of the original string.
def replace(self, old, new, count=-1, /)-
Return a copy with all occurrences of substring old replaced by new.
count Maximum number of occurrences to replace. -1 (the default value) means replace all occurrences.
If the optional argument count is given, only the first count occurrences are replaced.
def rfind(...)-
S.rfind(sub[, start[, end]]) -> int
Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.
Return -1 on failure.
def rindex(...)-
S.rindex(sub[, start[, end]]) -> int
Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.
Raises ValueError when the substring is not found.
def rjust(self, width, fillchar=' ', /)-
Return a right-justified string of length width.
Padding is done using the specified fill character (default is a space).
def rpartition(self, sep, /)-
Partition the string into three parts using the given separator.
This will search for the separator in the string, starting at the end. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.
If the separator is not found, returns a 3-tuple containing two empty strings and the original string.
def rsplit(self, /, sep=None, maxsplit=-1)-
Return a list of the substrings in the string, using sep as the separator string.
sep The separator used to split the string.
When set to None (the default value), will split on any whitespace character (including \n \r \t \f and spaces) and will discard empty strings from the result.maxsplit Maximum number of splits. -1 (the default value) means no limit.
Splitting starts at the end of the string and works to the front.
def rstrip(self, chars=None, /)-
Return a copy of the string with trailing whitespace removed.
If chars is given and not None, remove characters in chars instead.
def split(self, /, sep=None, maxsplit=-1)-
Return a list of the substrings in the string, using sep as the separator string.
sep The separator used to split the string.
When set to None (the default value), will split on any whitespace character (including \n \r \t \f and spaces) and will discard empty strings from the result.maxsplit Maximum number of splits. -1 (the default value) means no limit.
Splitting starts at the front of the string and works to the end.
Note, str.split() is mainly useful for data that has been intentionally delimited. With natural text that includes punctuation, consider using the regular expression module.
def splitlines(self, /, keepends=False)-
Return a list of the lines in the string, breaking at line boundaries.
Line breaks are not included in the resulting list unless keepends is given and true.
def startswith(...)-
S.startswith(prefix[, start[, end]]) -> bool
Return True if S starts with the specified prefix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. prefix can also be a tuple of strings to try.
def strip(self, chars=None, /)-
Return a copy of the string with leading and trailing whitespace removed.
If chars is given and not None, remove characters in chars instead.
def swapcase(self, /)-
Convert uppercase characters to lowercase and lowercase characters to uppercase.
def title(self, /)-
Return a version of the string where each word is titlecased.
More specifically, words start with uppercased characters and all remaining cased characters have lower case.
def translate(self, table, /)-
Replace each character in the string using the given translation table.
table Translation table, which must be a mapping of Unicode ordinals to Unicode ordinals, strings, or None.
The table must implement lookup/indexing via getitem, for instance a dictionary or list. If this operation raises LookupError, the character is left untouched. Characters mapped to None are deleted.
def upper(self, /)-
Return a copy of the string converted to uppercase.
def zfill(self, width, /)-
Pad a numeric string with zeros on the left, to fill a field of the given width.
The string is never truncated.