Module adcp.server.idempotency

Server-side idempotency middleware for AdCP mutating tool handlers.

Implements the seller side of AdCP #2315: extract idempotency_key, look up cached responses scoped by (authenticated_principal, idempotency_key), and replay the cached response verbatim when a subsequent request carries the same key + canonicalized-equivalent payload. Reject key reuse with a different payload as IDEMPOTENCY_CONFLICT.

The spec contract lives at adcontextprotocol/adcp/docs/building/implementation/security.mdx#idempotency.

Typical usage::

from adcp.server import ADCPHandler, IdempotencyStore, MemoryBackend, ToolContext
from adcp.server.responses import capabilities_response

idempotency = IdempotencyStore(
    backend=MemoryBackend(),
    ttl_seconds=86400,  # 24 hours, matches spec minimum
)

class MySeller(ADCPHandler):
    @idempotency.wrap
    async def create_media_buy(self, params, context=None):
        # <code>params</code> carries <code>idempotency\_key</code>; <code>context.caller\_identity</code>
        # identifies the principal. Without either, the middleware falls
        # through to this handler with no dedup (schema validation above
        # us rejects missing keys per AdCP #2315).
        return my_create_logic(params)

    async def get_adcp_capabilities(self, params, context=None):
        return capabilities_response(
            ["media_buy"],
            idempotency=idempotency.capability(),
        )

Callers who invoke the handler directly (tests, non-HTTP code paths) must pass a :class:ToolContext with caller_identity set — it's how the middleware scopes the cache namespace per-principal::

ctx = ToolContext(caller_identity="buyer-acme")
result = await seller.create_media_buy(
    {"idempotency_key": key, ...}, ctx
)

Backends:

• :class:MemoryBackend — in-process dict with TTL; use for tests and single-process reference implementations.
• :class:PgBackend — Postgres-backed store for multi-worker durable replay. Requires the adcp[pg] extra. await backend.create_schema() once at boot; commits go through a fresh pool connection (separate from the handler's transaction in v1 — co-tx wiring is a v1.1 affordance).

Sub-modules

adcp.server.idempotency.backends

Storage backends for :class:~adcp.server.idempotency.IdempotencyStore …

adcp.server.idempotency.canonicalize

Canonical payload hashing for AdCP idempotency replay detection …

adcp.server.idempotency.lazy

Deferred-construction wrapper for :class:IdempotencyBackend …

adcp.server.idempotency.store

The :class:IdempotencyStore coordinator: canonical hashing + backend + decorator …

adcp.server.idempotency.webhook_dedup

Webhook receiver-side dedup store …

Functions

def canonical_json_sha256(payload: dict[str, Any]) ‑> str

Expand source code

def canonical_json_sha256(payload: dict[str, Any]) -> str:
    """Compute the spec's canonical payload fingerprint.

    1. Strip the spec's exclusion list (see :func:`strip_excluded_fields`).
    2. RFC 8785 JCS canonicalize (stable key order, compact, UTF-8, spec-
       compliant number serialization).
    3. SHA-256 over the canonical bytes; return hex digest.

    The result is stable across all conforming JCS implementations. Two
    payloads whose hashes match are equivalent under AdCP replay semantics;
    two with different hashes are distinct and MUST be treated as a conflict
    when the caller supplies the same ``idempotency_key``.
    """
    stripped = strip_excluded_fields(payload)
    canonical = rfc8785.dumps(stripped)
    return hashlib.sha256(canonical).hexdigest()

Compute the spec's canonical payload fingerprint.

1. Strip the spec's exclusion list (see :func:strip_excluded_fields()).
2. RFC 8785 JCS canonicalize (stable key order, compact, UTF-8, spec- compliant number serialization).
3. SHA-256 over the canonical bytes; return hex digest.

The result is stable across all conforming JCS implementations. Two payloads whose hashes match are equivalent under AdCP replay semantics; two with different hashes are distinct and MUST be treated as a conflict when the caller supplies the same idempotency_key.

def create_lazy_backend(factory: LazyBackendFactory, *, allow_clear_all: bool = False) ‑> LazyBackend

Expand source code

def create_lazy_backend(
    factory: LazyBackendFactory,
    *,
    allow_clear_all: bool = False,
) -> LazyBackend:
    """Construct a :class:`LazyBackend` — functional alias mirroring the JS
    ``createLazyBackend`` factory shape.

    See :class:`LazyBackend` for semantics and parameter documentation.
    """
    return LazyBackend(factory, allow_clear_all=allow_clear_all)

Construct a :class:LazyBackend — functional alias mirroring the JS createLazyBackend factory shape.

See :class:LazyBackend for semantics and parameter documentation.

def is_wrapped(fn: Any) ‑> bool

Expand source code

def is_wrapped(fn: Any) -> bool:
    """Return True if ``fn`` was produced by :meth:`IdempotencyStore.wrap`.

    Accepts bound methods (resolves to the underlying function before
    the membership check) and plain callables. Used by the boot-time
    validator at :mod:`adcp.decisioning.validate_idempotency`.
    """
    if fn is None:
        return False
    target = fn.__func__ if hasattr(fn, "__func__") else fn
    return target in _WRAPPED_FUNCTIONS

Return True if fn was produced by :meth:IdempotencyStore.wrap().

Accepts bound methods (resolves to the underlying function before the membership check) and plain callables. Used by the boot-time validator at :mod:adcp.decisioning.validate_idempotency.

def strip_excluded_fields(payload: dict[str, Any]) ‑> dict[str, typing.Any]

Expand source code

def strip_excluded_fields(payload: dict[str, Any]) -> dict[str, Any]:
    """Return a deep copy of ``payload`` with the spec's exclusion list removed.

    Top-level keys in :data:`EXCLUDED_FIELDS` are dropped. Nested paths in
    ``_NESTED_EXCLUSIONS`` are traversed; the final leaf key is removed if the
    traversal reaches it. Missing intermediate keys are a no-op — the caller's
    payload is free to omit the push_notification_config entirely.

    The input dict is never mutated.
    """
    out: dict[str, Any] = copy.deepcopy(payload)
    for key in EXCLUDED_FIELDS:
        out.pop(key, None)
    for path in _NESTED_EXCLUSIONS:
        _drop_nested(out, path)
    return out

Return a deep copy of payload with the spec's exclusion list removed.

Top-level keys in :data:EXCLUDED_FIELDS are dropped. Nested paths in _NESTED_EXCLUSIONS are traversed; the final leaf key is removed if the traversal reaches it. Missing intermediate keys are a no-op — the caller's payload is free to omit the push_notification_config entirely.

The input dict is never mutated.

Classes

class CachedResponse (payload_hash: str, response: dict[str, Any], expires_at_epoch: float)

Expand source code

@dataclass(frozen=True)
class CachedResponse:
    """A single cached handler response keyed by ``(principal_id, key)``.

    :param payload_hash: Canonical JSON SHA-256 of the *original* request. On
        replay we compare the new request's hash to this value; mismatch is
        ``IDEMPOTENCY_CONFLICT``.
    :param response: The response dict the handler returned. On replay,
        :meth:`IdempotencyStore.wrap` injects ``replayed: true`` at the
        envelope level per AdCP L1/security idempotency rule 4 before
        returning to the caller — the cached value here stays clean so
        the same entry can serve multiple replays without compounding.
    :param expires_at_epoch: Unix timestamp (seconds) when this entry becomes
        eligible for eviction. Reads after this time return None.
    """

    payload_hash: str
    response: dict[str, Any]
    expires_at_epoch: float

A single cached handler response keyed by (principal_id, key).

:param payload_hash: Canonical JSON SHA-256 of the original request. On replay we compare the new request's hash to this value; mismatch is IDEMPOTENCY_CONFLICT. :param response: The response dict the handler returned. On replay, :meth:IdempotencyStore.wrap() injects replayed: true at the envelope level per AdCP L1/security idempotency rule 4 before returning to the caller — the cached value here stays clean so the same entry can serve multiple replays without compounding. :param expires_at_epoch: Unix timestamp (seconds) when this entry becomes eligible for eviction. Reads after this time return None.

Instance variables

var expires_at_epoch : float

var payload_hash : str

var response : dict[str, typing.Any]

class IdempotencyBackend

Expand source code

class IdempotencyBackend(ABC):
    """Abstract storage backend contract.

    All methods are async. Implementations MUST be safe to call concurrently
    from multiple asyncio tasks — :class:`IdempotencyStore` does not serialize
    access on the caller's behalf.
    """

    @abstractmethod
    async def get(self, scope_key: str, key: str) -> CachedResponse | None:
        """Return the cached entry, or None if missing or expired.

        ``scope_key`` is the caller-composed identity scope — typically
        ``tenant_id + caller_identity``. Backends treat it as an opaque
        string; the composition is owned by
        :class:`~adcp.server.idempotency.IdempotencyStore`.
        """

    @abstractmethod
    async def put(
        self,
        scope_key: str,
        key: str,
        entry: CachedResponse,
    ) -> None:
        """Store ``entry`` under ``(scope_key, key)``. Overwrites any prior
        entry — the store only calls ``put`` after verifying the slot is empty
        or expired, so an overwrite in that window is a legitimate retry of
        the write itself."""

    @abstractmethod
    async def delete_expired(self, now_epoch: float | None = None) -> int:
        """Best-effort sweep of expired entries. Returns the count removed.

        Sweeping is optional — :meth:`get` MUST self-filter expired entries.
        Backends that have natural TTL primitives (Redis ``EXPIRE``, Postgres
        partial indexes) may implement this as a no-op."""

Abstract storage backend contract.

All methods are async. Implementations MUST be safe to call concurrently from multiple asyncio tasks — :class:IdempotencyStore does not serialize access on the caller's behalf.

Ancestors

• abc.ABC

Subclasses

• MemoryBackend
• PgBackend
• LazyBackend

Methods

async def delete_expired(self, now_epoch: float | None = None) ‑> int

Expand source code

@abstractmethod
async def delete_expired(self, now_epoch: float | None = None) -> int:
    """Best-effort sweep of expired entries. Returns the count removed.

    Sweeping is optional — :meth:`get` MUST self-filter expired entries.
    Backends that have natural TTL primitives (Redis ``EXPIRE``, Postgres
    partial indexes) may implement this as a no-op."""

Best-effort sweep of expired entries. Returns the count removed.

Sweeping is optional — :meth:get MUST self-filter expired entries. Backends that have natural TTL primitives (Redis EXPIRE, Postgres partial indexes) may implement this as a no-op.

async def get(self, scope_key: str, key: str) ‑> CachedResponse | None

Expand source code

@abstractmethod
async def get(self, scope_key: str, key: str) -> CachedResponse | None:
    """Return the cached entry, or None if missing or expired.

    ``scope_key`` is the caller-composed identity scope — typically
    ``tenant_id + caller_identity``. Backends treat it as an opaque
    string; the composition is owned by
    :class:`~adcp.server.idempotency.IdempotencyStore`.
    """

Return the cached entry, or None if missing or expired.

scope_key is the caller-composed identity scope — typically tenant_id + caller_identity. Backends treat it as an opaque string; the composition is owned by :class:~adcp.server.idempotency.IdempotencyStore.

async def put(self,
scope_key: str,
key: str,
entry: CachedResponse) ‑> None

Expand source code

@abstractmethod
async def put(
    self,
    scope_key: str,
    key: str,
    entry: CachedResponse,
) -> None:
    """Store ``entry`` under ``(scope_key, key)``. Overwrites any prior
    entry — the store only calls ``put`` after verifying the slot is empty
    or expired, so an overwrite in that window is a legitimate retry of
    the write itself."""

Store entry under (scope_key, key). Overwrites any prior entry — the store only calls put after verifying the slot is empty or expired, so an overwrite in that window is a legitimate retry of the write itself.

class IdempotencyStore (backend: IdempotencyBackend,
ttl_seconds: int = 86400,
hash_fn: Callable[[dict[str, Any]], str] = <function canonical_json_sha256>,
*,
clock: Callable[[], float] = <built-in function time>)

Expand source code

class IdempotencyStore:
    """Coordinator that binds canonical hashing to a storage backend.

    :param backend: A concrete :class:`IdempotencyBackend`.
    :param ttl_seconds: How long cached responses remain replayable. Must be
        within the spec's ``[3600, 604800]`` range (1h to 7d). 86400 (24h) is
        the recommended floor and matches the compliance storyboard.
    :param hash_fn: Optional override for the canonical hash function. Defaults
        to :func:`canonical_json_sha256`. Exposed for tests and for anyone who
        wants to experiment with alternative equivalence rules — though note
        the spec mandates RFC 8785 JCS for interop.
    """

    def __init__(
        self,
        backend: IdempotencyBackend,
        ttl_seconds: int = 86400,
        hash_fn: Callable[[dict[str, Any]], str] = canonical_json_sha256,
        *,
        clock: Callable[[], float] = time.time,
    ) -> None:
        if not _MIN_TTL_SECONDS <= ttl_seconds <= _MAX_TTL_SECONDS:
            raise ValueError(
                f"ttl_seconds must be in [{_MIN_TTL_SECONDS}, {_MAX_TTL_SECONDS}] "
                f"per AdCP spec (capabilities.idempotency.replay_ttl_seconds), "
                f"got {ttl_seconds}"
            )
        self.backend = backend
        self.ttl_seconds = ttl_seconds
        self._hash_fn = hash_fn
        self._clock = clock

    def capability(self) -> dict[str, Any]:
        """Return the capabilities fragment declaring this store's replay window.

        Embed under ``capabilities.adcp.idempotency`` on the seller's
        ``get_adcp_capabilities`` response. Buyers read this to reason about
        retry-safe windows (AdCP #2315)::

            caps.adcp.idempotency = idempotency.capability()
            # → {"supported": True, "replay_ttl_seconds": 86400}

        ``supported`` became REQUIRED in AdCP 3.0 GA — agents emitting only
        ``replay_ttl_seconds`` fail strict schema validation on the new
        capabilities response.
        """
        return {"supported": True, "replay_ttl_seconds": self.ttl_seconds}

    def wrap(self, handler: HandlerFn) -> HandlerFn:
        """Decorator that adds idempotency semantics to an AdCP handler method.

        Supports three calling conventions the framework dispatches with:

        1. **Positional** ``handler(self, params, context)`` — the
           default for non-projected tools (``get_products``,
           ``create_media_buy``, etc.).
        2. **Keyword** ``handler(self, params=..., context=...)`` —
           same shape, just kwargs.
        3. **Arg-projected** ``handler(self, **arg_projector_kwargs, ctx=...)``
           where ``params`` is split into per-field kwargs by the
           framework dispatcher (e.g. ``update_media_buy`` is called
           as ``handler(self, media_buy_id=..., patch=..., ctx=...)``).
           In this mode the wrap searches the kwargs for a Pydantic
           model (``patch`` for update_media_buy) to extract the
           idempotency key and hash payload from. Adopters whose
           projection contains no Pydantic model (e.g. a method
           projecting only a list of ids) get fall-through behavior:
           no key found → handler runs without dedup.

        ``params`` is normalized to a dict before hashing; the return
        value is coerced to a dict for caching (via ``model_dump`` if
        Pydantic). The decorator always returns the handler's original
        object on a cache miss and a best-effort Pydantic
        re-validation on a hit (when the handler's declared return
        type exposes ``model_validate``). Callers that return raw
        dicts get dicts back.
        """

        @wraps(handler)
        async def _wrapped(*args: Any, **kwargs: Any) -> Any:
            handler_self, hash_source, context = _resolve_call_args(args, kwargs)

            scope_key, idempotency_key, params_dict = self._prepare(hash_source, context)
            if scope_key is None or idempotency_key is None:
                # No key → spec says the server MUST reject with INVALID_REQUEST.
                # We let the handler run so validation layers above us (Pydantic,
                # FastAPI, etc.) can reject with a typed error; the middleware's
                # job is only to dedup when a key IS present.
                #
                # Forward the call exactly as received so all three calling
                # conventions (positional / keyword / arg-projected) reach
                # the inner handler unchanged. The wrap is signature-
                # transparent on the no-key path.
                return await handler(*args, **kwargs)

            payload_hash = self._hash_fn(params_dict)

            cached = await self.backend.get(scope_key, idempotency_key)
            if cached is not None:
                if cached.payload_hash == payload_hash:
                    logger.debug(
                        "idempotency replay: scope=%s key_prefix=%s",
                        _scope_log_id(scope_key),
                        idempotency_key[:8],
                    )
                    # AdCP L1/security idempotency rule 4: the replay
                    # envelope MUST carry ``replayed: true`` so buyer
                    # agents can suppress side effects (notifications,
                    # webhook dispatch, memory writes) on retry. The
                    # store owns this — sellers can't inject at the
                    # right point (cache lookup happens here, wire
                    # serialization happens later). The injection
                    # lands on the cloned dict, not ``cached.response``,
                    # so multiple replays of the same key all carry
                    # exactly one ``replayed: true`` without compounding.
                    replay = _clone_response(cached.response)
                    replay["replayed"] = True
                    return replay
                # Same key, different payload — spec-defined conflict.
                raise IdempotencyConflictError(
                    operation=getattr(handler, "__name__", "handler"),
                    errors=[
                        {
                            "code": "IDEMPOTENCY_CONFLICT",
                            "message": (
                                "idempotency_key reused with a different payload "
                                "(canonical hash mismatch)"
                            ),
                        }
                    ],
                )

            response = await handler(*args, **kwargs)
            # Deep-copy when caching so post-return mutation of the caller's
            # copy can't poison future replays. `_clone_response` also deep-
            # copies on the hit path, giving independent objects per replay.
            response_dict = copy.deepcopy(_to_dict(response))
            entry = CachedResponse(
                payload_hash=payload_hash,
                response=response_dict,
                expires_at_epoch=self._clock() + self.ttl_seconds,
            )
            # Commit cache AFTER handler returns. Atomicity with the handler's
            # side effects depends on the backend: MemoryBackend is best-effort
            # (no transactional relationship to external resources); PgBackend
            # (follow-up) will commit in the same transaction when the handler
            # uses the same engine. On put failure we log loudly and return
            # the handler's response — swallowing the exception would be wrong
            # (operators need the signal that caching is broken), and raising
            # would look to the caller like the handler failed, triggering a
            # retry that re-executes side effects. Best compromise: warn
            # operators, return the result, and accept that the next retry
            # with this key will re-execute.
            try:
                await self.backend.put(scope_key, idempotency_key, entry)
            except Exception:
                logger.warning(
                    "Idempotency cache put failed for scope=%s key_prefix=%s — "
                    "handler completed but a subsequent retry with this key will "
                    "re-execute rather than replay. This indicates an operational "
                    "issue with the idempotency backend.",
                    _scope_log_id(scope_key),
                    idempotency_key[:8],
                    exc_info=True,
                )
            return response

        # Register the wrapper for the boot-time validator at
        # adcp.decisioning.validate_idempotency. WeakSet membership —
        # not a public attribute — so adopters can't spoof "wrapped"
        # by stamping an attr on a plain function. The wrapper is
        # registered, not the original handler: re-decorating a forked
        # copy of `handler` would otherwise falsely flag both.
        #
        # Contract for future maintainers: ``is_wrapped()`` checks
        # WeakSet membership of the closure object directly. Do NOT
        # change it to ``inspect.unwrap()``-then-check — the
        # ``@functools.wraps(handler)`` decorator above sets
        # ``_wrapped.__wrapped__ = handler``, so ``inspect.unwrap``
        # would return the original handler (not in the WeakSet) and
        # the validator would silently regress.
        _WRAPPED_FUNCTIONS.add(_wrapped)
        return _wrapped

    def _prepare(self, params: Any, context: Any) -> tuple[str | None, str | None, dict[str, Any]]:
        """Normalize inputs and extract the (scope_key, key, params_dict) tuple.

        ``scope_key`` composes ``tenant_id`` (when present) with
        ``caller_identity`` so cache entries are isolated across tenants even
        if the seller's principal IDs are only unique within each tenant.

        Returns ``(None, None, params_dict)`` when idempotency doesn't apply
        (no caller identity or no key supplied). The caller falls through to
        the plain handler in that case — validation of missing-key lives in
        the request schema, not here.
        """
        params_dict = _to_dict(params)
        idempotency_key = params_dict.get("idempotency_key")
        if not isinstance(idempotency_key, str) or not idempotency_key:
            return None, None, params_dict
        scope_key = _extract_scope_key(context)
        if scope_key is None:
            # No caller identity: we can't safely scope the key. Spec requires
            # per-principal scope; anything else is a cross-principal replay
            # attack surface. Fall through to the handler (which will process
            # the request normally — no dedup, but no security regression).
            self._warn_missing_principal_once()
            return None, None, params_dict
        return scope_key, idempotency_key, params_dict

    _missing_principal_warned: bool = False

    def _warn_missing_principal_once(self) -> None:
        """Emit a one-time warning when the middleware sees a key but no principal.

        Silent fall-through is the worst DX: the seller drops in
        ``@idempotency.wrap``, ships, and doesn't discover until incident
        review that no dedup ever happened. Fire once per store instance so
        operators see the signal without filling logs on every request.
        """
        if self._missing_principal_warned:
            return
        self._missing_principal_warned = True
        warnings.warn(
            "IdempotencyStore received a request with idempotency_key but no "
            "caller_identity on ToolContext — dedup is SKIPPED. This usually "
            "means your transport isn't populating the authenticated principal. "
            "A2A: wire an a2a-sdk auth middleware that sets ServerCallContext.user; "
            "MCP: populate ToolContext.caller_identity from your FastMCP auth "
            "middleware (see adcp.server.idempotency README). "
            "This warning fires once per IdempotencyStore instance.",
            UserWarning,
            stacklevel=3,
        )

Coordinator that binds canonical hashing to a storage backend.

:param backend: A concrete :class:IdempotencyBackend. :param ttl_seconds: How long cached responses remain replayable. Must be within the spec's [3600, 604800] range (1h to 7d). 86400 (24h) is the recommended floor and matches the compliance storyboard. :param hash_fn: Optional override for the canonical hash function. Defaults to :func:canonical_json_sha256(). Exposed for tests and for anyone who wants to experiment with alternative equivalence rules — though note the spec mandates RFC 8785 JCS for interop.

Methods

def capability(self) ‑> dict[str, typing.Any]

Expand source code

def capability(self) -> dict[str, Any]:
    """Return the capabilities fragment declaring this store's replay window.

    Embed under ``capabilities.adcp.idempotency`` on the seller's
    ``get_adcp_capabilities`` response. Buyers read this to reason about
    retry-safe windows (AdCP #2315)::

        caps.adcp.idempotency = idempotency.capability()
        # → {"supported": True, "replay_ttl_seconds": 86400}

    ``supported`` became REQUIRED in AdCP 3.0 GA — agents emitting only
    ``replay_ttl_seconds`` fail strict schema validation on the new
    capabilities response.
    """
    return {"supported": True, "replay_ttl_seconds": self.ttl_seconds}

Return the capabilities fragment declaring this store's replay window.

Embed under capabilities.adcp.idempotency on the seller's get_adcp_capabilities response. Buyers read this to reason about retry-safe windows (AdCP #2315)::

caps.adcp.idempotency = idempotency.capability()
# → {"supported": True, "replay_ttl_seconds": 86400}

supported became REQUIRED in AdCP 3.0 GA — agents emitting only replay_ttl_seconds fail strict schema validation on the new capabilities response.

def wrap(self, handler: HandlerFn) ‑> Callable[..., Awaitable[typing.Any]]

Expand source code

def wrap(self, handler: HandlerFn) -> HandlerFn:
    """Decorator that adds idempotency semantics to an AdCP handler method.

    Supports three calling conventions the framework dispatches with:

    1. **Positional** ``handler(self, params, context)`` — the
       default for non-projected tools (``get_products``,
       ``create_media_buy``, etc.).
    2. **Keyword** ``handler(self, params=..., context=...)`` —
       same shape, just kwargs.
    3. **Arg-projected** ``handler(self, **arg_projector_kwargs, ctx=...)``
       where ``params`` is split into per-field kwargs by the
       framework dispatcher (e.g. ``update_media_buy`` is called
       as ``handler(self, media_buy_id=..., patch=..., ctx=...)``).
       In this mode the wrap searches the kwargs for a Pydantic
       model (``patch`` for update_media_buy) to extract the
       idempotency key and hash payload from. Adopters whose
       projection contains no Pydantic model (e.g. a method
       projecting only a list of ids) get fall-through behavior:
       no key found → handler runs without dedup.

    ``params`` is normalized to a dict before hashing; the return
    value is coerced to a dict for caching (via ``model_dump`` if
    Pydantic). The decorator always returns the handler's original
    object on a cache miss and a best-effort Pydantic
    re-validation on a hit (when the handler's declared return
    type exposes ``model_validate``). Callers that return raw
    dicts get dicts back.
    """

    @wraps(handler)
    async def _wrapped(*args: Any, **kwargs: Any) -> Any:
        handler_self, hash_source, context = _resolve_call_args(args, kwargs)

        scope_key, idempotency_key, params_dict = self._prepare(hash_source, context)
        if scope_key is None or idempotency_key is None:
            # No key → spec says the server MUST reject with INVALID_REQUEST.
            # We let the handler run so validation layers above us (Pydantic,
            # FastAPI, etc.) can reject with a typed error; the middleware's
            # job is only to dedup when a key IS present.
            #
            # Forward the call exactly as received so all three calling
            # conventions (positional / keyword / arg-projected) reach
            # the inner handler unchanged. The wrap is signature-
            # transparent on the no-key path.
            return await handler(*args, **kwargs)

        payload_hash = self._hash_fn(params_dict)

        cached = await self.backend.get(scope_key, idempotency_key)
        if cached is not None:
            if cached.payload_hash == payload_hash:
                logger.debug(
                    "idempotency replay: scope=%s key_prefix=%s",
                    _scope_log_id(scope_key),
                    idempotency_key[:8],
                )
                # AdCP L1/security idempotency rule 4: the replay
                # envelope MUST carry ``replayed: true`` so buyer
                # agents can suppress side effects (notifications,
                # webhook dispatch, memory writes) on retry. The
                # store owns this — sellers can't inject at the
                # right point (cache lookup happens here, wire
                # serialization happens later). The injection
                # lands on the cloned dict, not ``cached.response``,
                # so multiple replays of the same key all carry
                # exactly one ``replayed: true`` without compounding.
                replay = _clone_response(cached.response)
                replay["replayed"] = True
                return replay
            # Same key, different payload — spec-defined conflict.
            raise IdempotencyConflictError(
                operation=getattr(handler, "__name__", "handler"),
                errors=[
                    {
                        "code": "IDEMPOTENCY_CONFLICT",
                        "message": (
                            "idempotency_key reused with a different payload "
                            "(canonical hash mismatch)"
                        ),
                    }
                ],
            )

        response = await handler(*args, **kwargs)
        # Deep-copy when caching so post-return mutation of the caller's
        # copy can't poison future replays. `_clone_response` also deep-
        # copies on the hit path, giving independent objects per replay.
        response_dict = copy.deepcopy(_to_dict(response))
        entry = CachedResponse(
            payload_hash=payload_hash,
            response=response_dict,
            expires_at_epoch=self._clock() + self.ttl_seconds,
        )
        # Commit cache AFTER handler returns. Atomicity with the handler's
        # side effects depends on the backend: MemoryBackend is best-effort
        # (no transactional relationship to external resources); PgBackend
        # (follow-up) will commit in the same transaction when the handler
        # uses the same engine. On put failure we log loudly and return
        # the handler's response — swallowing the exception would be wrong
        # (operators need the signal that caching is broken), and raising
        # would look to the caller like the handler failed, triggering a
        # retry that re-executes side effects. Best compromise: warn
        # operators, return the result, and accept that the next retry
        # with this key will re-execute.
        try:
            await self.backend.put(scope_key, idempotency_key, entry)
        except Exception:
            logger.warning(
                "Idempotency cache put failed for scope=%s key_prefix=%s — "
                "handler completed but a subsequent retry with this key will "
                "re-execute rather than replay. This indicates an operational "
                "issue with the idempotency backend.",
                _scope_log_id(scope_key),
                idempotency_key[:8],
                exc_info=True,
            )
        return response

    # Register the wrapper for the boot-time validator at
    # adcp.decisioning.validate_idempotency. WeakSet membership —
    # not a public attribute — so adopters can't spoof "wrapped"
    # by stamping an attr on a plain function. The wrapper is
    # registered, not the original handler: re-decorating a forked
    # copy of `handler` would otherwise falsely flag both.
    #
    # Contract for future maintainers: ``is_wrapped()`` checks
    # WeakSet membership of the closure object directly. Do NOT
    # change it to ``inspect.unwrap()``-then-check — the
    # ``@functools.wraps(handler)`` decorator above sets
    # ``_wrapped.__wrapped__ = handler``, so ``inspect.unwrap``
    # would return the original handler (not in the WeakSet) and
    # the validator would silently regress.
    _WRAPPED_FUNCTIONS.add(_wrapped)
    return _wrapped

Decorator that adds idempotency semantics to an AdCP handler method.

Supports three calling conventions the framework dispatches with:

1. Positional handler(self, params, context) — the default for non-projected tools (get_products, create_media_buy, etc.).
2. Keyword handler(self, params=..., context=...) — same shape, just kwargs.
3. Arg-projected handler(self, **arg_projector_kwargs, ctx=...) where params is split into per-field kwargs by the framework dispatcher (e.g. update_media_buy is called as handler(self, media_buy_id=..., patch=..., ctx=...)). In this mode the wrap searches the kwargs for a Pydantic model (patch for update_media_buy) to extract the idempotency key and hash payload from. Adopters whose projection contains no Pydantic model (e.g. a method projecting only a list of ids) get fall-through behavior: no key found → handler runs without dedup.

params is normalized to a dict before hashing; the return value is coerced to a dict for caching (via model_dump if Pydantic). The decorator always returns the handler's original object on a cache miss and a best-effort Pydantic re-validation on a hit (when the handler's declared return type exposes model_validate). Callers that return raw dicts get dicts back.

class LazyBackend (factory: LazyBackendFactory, *, allow_clear_all: bool = False)

Expand source code

class LazyBackend(IdempotencyBackend):
    """Resolve an :class:`IdempotencyBackend` lazily on first use.

    :param factory: A zero-arg callable returning an :class:`IdempotencyBackend`
        or an awaitable that resolves to one. Invoked at most once across the
        wrapper's lifetime once it succeeds; re-invoked only if a prior attempt
        raised.
    :param allow_clear_all: When ``True``, expose :meth:`clear_all`, delegating
        to the resolved backend's ``clear_all`` or ``clear`` method. Defaults to
        ``False`` because bulk clearing is dangerous on shared production stores
        and the SDK uses method presence as the reset-safety contract.

    Concurrency: the first ``get``/``put``/``delete_expired`` triggers
    resolution. Multiple concurrent first operations share a single factory
    invocation via an :class:`asyncio.Lock`; later callers reuse the cached
    instance without locking on the hot path.
    """

    def __init__(
        self,
        factory: LazyBackendFactory,
        *,
        allow_clear_all: bool = False,
    ) -> None:
        self._factory = factory
        self._allow_clear_all = allow_clear_all
        self._backend: IdempotencyBackend | None = None
        self._lock = asyncio.Lock()

    async def _resolve(self) -> IdempotencyBackend:
        """Return the resolved backend, invoking the factory once on first use.

        Resolve-once + concurrency-safe: the fast path returns the memoized
        instance without locking. The slow path holds ``_lock`` so concurrent
        first callers share a single factory invocation; the double-check inside
        the lock means a caller that waited on the lock sees the instance the
        winner resolved. A factory that raises is not memoized — the next call
        retries.
        """
        cached = self._backend
        if cached is not None:
            return cached
        async with self._lock:
            # Re-read under the lock: a task that lost the race to acquire it
            # must observe the winner's resolved instance, not re-run the
            # factory. (Read into a local so the narrowing is on the local,
            # not the instance attribute another task may have mutated.)
            cached = self._backend
            if cached is not None:
                return cached
            result = self._factory()
            resolved = await result if isinstance(result, Awaitable) else result
            if not isinstance(resolved, IdempotencyBackend):
                raise TypeError(
                    "LazyBackend factory must resolve to an IdempotencyBackend, "
                    f"got {type(resolved).__name__}"
                )
            self._backend = resolved
            return resolved

    async def get(self, scope_key: str, key: str) -> CachedResponse | None:
        return await (await self._resolve()).get(scope_key, key)

    async def put(self, scope_key: str, key: str, entry: CachedResponse) -> None:
        await (await self._resolve()).put(scope_key, key, entry)

    async def delete_expired(self, now_epoch: float | None = None) -> int:
        return await (await self._resolve()).delete_expired(now_epoch)

    async def _clear_all(self) -> None:
        """Delegate a bulk clear to the resolved backend.

        Resolves the backend (so the factory runs if it hasn't yet) and
        delegates to its ``clear_all`` or ``clear`` method, raising if the
        resolved backend supports neither. Exposed as ``clear_all`` only when
        the wrapper is constructed with ``allow_clear_all=True`` (see
        :meth:`__getattr__`).
        """
        backend = await self._resolve()
        clear = getattr(backend, "clear_all", None) or getattr(backend, "clear", None)
        if clear is None:
            raise NotImplementedError(
                f"Resolved backend {type(backend).__name__} does not support "
                "clear_all() or clear()."
            )
        await clear()

    def __getattr__(self, name: str) -> object:
        """Expose ``clear_all`` only when opted in.

        Mirrors the JS wrapper, which attaches ``clearAll`` to the returned
        object solely when ``{ clearAll: true }`` — reset-safety code uses
        ``hasattr``/method presence as the "safe to flush" contract, so the
        attribute must genuinely be absent otherwise. ``__getattr__`` is only
        consulted for names not found normally, so this never shadows the
        delegating methods above.
        """
        if name == "clear_all" and self.__dict__.get("_allow_clear_all"):
            return self._clear_all
        raise AttributeError(f"{type(self).__name__!r} object has no attribute {name!r}")

Resolve an :class:IdempotencyBackend lazily on first use.

:param factory: A zero-arg callable returning an :class:IdempotencyBackend or an awaitable that resolves to one. Invoked at most once across the wrapper's lifetime once it succeeds; re-invoked only if a prior attempt raised. :param allow_clear_all: When True, expose :meth:clear_all, delegating to the resolved backend's clear_all or clear method. Defaults to False because bulk clearing is dangerous on shared production stores and the SDK uses method presence as the reset-safety contract.

Concurrency: the first get/put/delete_expired triggers resolution. Multiple concurrent first operations share a single factory invocation via an :class:asyncio.Lock; later callers reuse the cached instance without locking on the hot path.

Ancestors

• IdempotencyBackend
• abc.ABC

Inherited members

• IdempotencyBackend:
  • delete_expired
  • get
  • put

class MemoryBackend (*, clock: Callable[[], float] = <built-in function time>)

Expand source code

class MemoryBackend(IdempotencyBackend):
    """In-process dict-backed store.

    Suitable for tests, single-process reference implementations, and local
    development. **Not suitable for multi-process deployments** — each worker
    has its own cache, so a retry that lands on a different worker is treated
    as a fresh request.

    Thread safety: the backend uses an :class:`asyncio.Lock` to serialize
    mutations of the shared dict. Reads go through the lock too; for a pure
    in-process backend this is cheap and prevents torn reads across concurrent
    ``get``/``put`` interleaving.

    :param clock: Callable returning the current epoch seconds. Override for
        tests that need to advance time deterministically without monkeypatching
        :mod:`time`. Defaults to :func:`time.time`.
    """

    def __init__(self, *, clock: Callable[[], float] = time.time) -> None:
        self._store: dict[tuple[str, str], CachedResponse] = {}
        self._lock = asyncio.Lock()
        self._clock = clock

    async def get(self, scope_key: str, key: str) -> CachedResponse | None:
        async with self._lock:
            entry = self._store.get((scope_key, key))
            if entry is None:
                return None
            if entry.expires_at_epoch <= self._clock():
                # Lazy expiry — drop the stale entry so the next request
                # treats the slot as fresh and races to repopulate.
                del self._store[(scope_key, key)]
                return None
            return entry

    async def put(
        self,
        scope_key: str,
        key: str,
        entry: CachedResponse,
    ) -> None:
        async with self._lock:
            self._store[(scope_key, key)] = entry

    async def delete_expired(self, now_epoch: float | None = None) -> int:
        cutoff = now_epoch if now_epoch is not None else self._clock()
        async with self._lock:
            stale = [k for k, v in self._store.items() if v.expires_at_epoch <= cutoff]
            for k in stale:
                del self._store[k]
            return len(stale)

    async def clear(self) -> None:
        """Remove all cached entries.

        Test-suite hook — handy for resetting state between fixtures when a
        single :class:`MemoryBackend` is shared across multiple tests.
        """
        async with self._lock:
            self._store.clear()

    async def _size(self) -> int:
        """Test-only: return the current entry count."""
        async with self._lock:
            return len(self._store)

In-process dict-backed store.

Suitable for tests, single-process reference implementations, and local development. Not suitable for multi-process deployments — each worker has its own cache, so a retry that lands on a different worker is treated as a fresh request.

Thread safety: the backend uses an :class:asyncio.Lock to serialize mutations of the shared dict. Reads go through the lock too; for a pure in-process backend this is cheap and prevents torn reads across concurrent get/put interleaving.

:param clock: Callable returning the current epoch seconds. Override for tests that need to advance time deterministically without monkeypatching :mod:time. Defaults to :func:time.time.

Ancestors

• IdempotencyBackend
• abc.ABC

Methods

async def clear(self) ‑> None

Expand source code

async def clear(self) -> None:
    """Remove all cached entries.

    Test-suite hook — handy for resetting state between fixtures when a
    single :class:`MemoryBackend` is shared across multiple tests.
    """
    async with self._lock:
        self._store.clear()

Remove all cached entries.

Test-suite hook — handy for resetting state between fixtures when a single :class:MemoryBackend is shared across multiple tests.

Inherited members

• IdempotencyBackend:
  • delete_expired
  • get
  • put

class PgBackend (*, pool: Any, table_name: str = 'adcp_idempotency')

Expand source code

class PgBackend(IdempotencyBackend):
    """PostgreSQL-backed :class:`IdempotencyBackend`.

    Multi-worker durable replay cache. Adopters running ≥2 processes wire
    this in place of :class:`MemoryBackend` so a retry that lands on a
    different worker still replays the cached response.

    Example::

        from psycopg_pool import AsyncConnectionPool
        from adcp.server.idempotency import IdempotencyStore, PgBackend

        pool = AsyncConnectionPool("postgresql://...", min_size=2, max_size=10)
        backend = PgBackend(pool=pool)
        await backend.create_schema()  # idempotent; safe to call on every boot

        store = IdempotencyStore(backend=backend, ttl_seconds=86400)

    **Atomicity caveat (v1).** ``put`` commits on a fresh pool connection —
    the cache write is NOT in the same transaction as the handler's
    business writes. A crash between handler success and cache commit
    leaves the slot empty; the next retry re-executes the handler.
    Idempotent handlers absorb this without harm. **Handlers with
    non-idempotent side effects** (e.g., ``INSERT INTO media_buys``
    without a unique constraint on the buyer's idempotency_key) need
    either: (a) handler-level dedupe via a database unique constraint
    that maps to the same key the SDK uses, or (b) the co-tx variant
    once it ships. Co-tx — handler passes its own connection so the
    cache write commits atomically with side effects — is planned as a
    follow-on enhancement.

    **Schema bootstrap caveat.** :meth:`create_schema` uses
    ``CREATE TABLE IF NOT EXISTS`` — if a table with the same name but
    a different shape already exists (Alembic migration drift, manual
    DDL with ``response JSON`` instead of ``JSONB``, missing
    ``COLLATE "C"``), this method is a no-op and the backend will run
    against the wrong column types. If you manage the schema with
    Alembic / dbmate, copy the DDL inside :meth:`create_schema`
    verbatim into a migration revision — keep ``COLLATE "C"`` and
    ``JSONB`` identical — and skip calling :meth:`create_schema` at
    boot.

    **Response payload contract.** :attr:`CachedResponse.response` is
    serialized via ``json.dumps`` for the JSONB column. Values must be
    JSON-safe — no ``datetime``, ``Decimal``, ``set``, or ``bytes``.
    Coerce in your handler before returning.

    **Cardinality / DoS.** This backend has no row cap; only TTL
    bounds the table size. Per AdCP spec, per-principal rate limiting
    at the auth tier is required — the backend trusts that. Schedule
    :meth:`delete_expired` as a cron / pg_cron / app-loop sweep
    (``get`` self-filters expired rows, but they accumulate on disk
    until something deletes them).

    **Schema.** Created idempotently by :meth:`create_schema`:

    .. code-block:: sql

        CREATE TABLE IF NOT EXISTS adcp_idempotency (
            scope_key    TEXT        COLLATE "C" NOT NULL,
            key          TEXT        COLLATE "C" NOT NULL,
            payload_hash TEXT        NOT NULL,
            response     JSONB       NOT NULL,
            expires_at   TIMESTAMPTZ NOT NULL,
            PRIMARY KEY (scope_key, key)
        );
        CREATE INDEX IF NOT EXISTS adcp_idempotency_expires_idx
            ON adcp_idempotency (expires_at);

    ``COLLATE "C"`` on identifier columns avoids locale-driven equivalence
    (``Principal-A`` ≡ ``principal-a`` under Turkish/locale-aware
    collations) collapsing distinct tenants into the same cache slot.

    :param pool: ``psycopg_pool.AsyncConnectionPool`` owned by the caller.
        Each operation acquires a short-lived connection. We don't open,
        own, or close the pool.
    :param table_name: Override the default table name. Useful for
        multi-tenant schema scoping. Default ``adcp_idempotency``.

    :raises ImportError: when psycopg/psycopg-pool are not installed.
        Install via the ``pg`` extra: ``pip install 'adcp[pg]'``.
    :raises ValueError: when ``table_name`` is not a safe ASCII
        identifier (``[a-z_][a-z0-9_]{0,62}``).
    """

    def __init__(
        self,
        *,
        pool: Any,  # psycopg_pool.AsyncConnectionPool — Any avoids runtime psycopg import
        table_name: str = DEFAULT_IDEMPOTENCY_TABLE,
    ) -> None:
        if not _PG_AVAILABLE:
            raise ImportError(_PG_INSTALL_HINT)
        self._pool = pool
        self._table = _safe_identifier(table_name)

        # Pre-format SQL once. Validated identifier so f-string interpolation
        # is byte-safe; values always go through %s parameterization. Same
        # convention as PgWebhookDeliverySupervisor / PgReplayStore.
        t = self._table
        self._sql_get = (
            f"SELECT payload_hash, response, expires_at "  # noqa: S608
            f"FROM {t} WHERE scope_key = %s AND key = %s AND expires_at > now()"
        )
        # First-writer-wins under concurrent put. The store's pre-check
        # ("slot is empty or expired") is NOT a lock — two workers can
        # both see an empty slot and race into put. With a naive
        # last-writer-wins ON CONFLICT, the second put would overwrite
        # the first's payload_hash, violating the cache invariant
        # "same (scope, key) → same hash". The WHERE on the UPDATE
        # arm restricts the overwrite to actually-expired rows: a
        # concurrent fresh write becomes a no-op, both callers
        # observe an equivalent cached entry from the first writer.
        self._sql_put = (
            f"INSERT INTO {t} "  # noqa: S608
            f"(scope_key, key, payload_hash, response, expires_at) "
            f"VALUES (%s, %s, %s, %s::jsonb, %s) "
            f"ON CONFLICT (scope_key, key) DO UPDATE SET "
            f"  payload_hash = EXCLUDED.payload_hash, "
            f"  response     = EXCLUDED.response, "
            f"  expires_at   = EXCLUDED.expires_at "
            f"WHERE {t}.expires_at <= now()"
        )
        self._sql_delete_expired = f"DELETE FROM {t} WHERE expires_at <= %s"  # noqa: S608

    async def create_schema(self) -> None:
        """Bootstrap the table + index. Idempotent.

        Safe to call on every app boot. Each DDL statement is executed
        separately — psycopg does not split on ``;``.
        """
        t = self._table
        statements = [
            f"""CREATE TABLE IF NOT EXISTS {t} (
                scope_key    TEXT        COLLATE "C" NOT NULL,
                key          TEXT        COLLATE "C" NOT NULL,
                payload_hash TEXT        NOT NULL,
                response     JSONB       NOT NULL,
                expires_at   TIMESTAMPTZ NOT NULL,
                PRIMARY KEY (scope_key, key)
            )""",
            # Partial-free expiry index — cheap eviction sweep.
            f"""CREATE INDEX IF NOT EXISTS {t}_expires_idx
                ON {t} (expires_at)""",
        ]
        async with self._pool.connection() as conn:
            for stmt in statements:
                await conn.execute(stmt)

    async def get(self, scope_key: str, key: str) -> CachedResponse | None:
        """Read the cached entry, filtering expired rows in the WHERE clause.

        Lazy expiry — expired rows stay on disk until ``delete_expired``
        sweeps them. ``get`` self-filters via ``expires_at > now()`` so a
        stale row never replays.
        """
        async with self._pool.connection() as conn:
            cur = await conn.execute(self._sql_get, (scope_key, key))
            row = await cur.fetchone()
            if row is None:
                return None
            payload_hash, response, expires_at = row
            return CachedResponse(
                payload_hash=payload_hash,
                response=response if isinstance(response, dict) else json.loads(response),
                expires_at_epoch=_to_epoch(expires_at),
            )

    async def put(
        self,
        scope_key: str,
        key: str,
        entry: CachedResponse,
    ) -> None:
        """Atomic upsert under ``(scope_key, key)``.

        ``ON CONFLICT DO UPDATE`` because the store only calls ``put``
        after verifying the slot is empty or expired — an overwrite in
        that window is a legitimate retry of the write itself.
        """
        expires_at_dt = datetime.fromtimestamp(entry.expires_at_epoch, tz=timezone.utc)
        async with self._pool.connection() as conn:
            await conn.execute(
                self._sql_put,
                (
                    scope_key,
                    key,
                    entry.payload_hash,
                    json.dumps(entry.response),
                    expires_at_dt,
                ),
            )

    async def delete_expired(self, now_epoch: float | None = None) -> int:
        """Best-effort sweep of expired entries. Returns rows removed."""
        cutoff = now_epoch if now_epoch is not None else time.time()
        cutoff_dt = datetime.fromtimestamp(cutoff, tz=timezone.utc)
        async with self._pool.connection() as conn:
            cur = await conn.execute(self._sql_delete_expired, (cutoff_dt,))
            return cur.rowcount or 0

PostgreSQL-backed :class:IdempotencyBackend.

Multi-worker durable replay cache. Adopters running ≥2 processes wire this in place of :class:MemoryBackend so a retry that lands on a different worker still replays the cached response.

Example::

from psycopg_pool import AsyncConnectionPool
from adcp.server.idempotency import IdempotencyStore, PgBackend

pool = AsyncConnectionPool("postgresql://...", min_size=2, max_size=10)
backend = PgBackend(pool=pool)
await backend.create_schema()  # idempotent; safe to call on every boot

store = IdempotencyStore(backend=backend, ttl_seconds=86400)

Atomicity caveat (v1). put commits on a fresh pool connection — the cache write is NOT in the same transaction as the handler's business writes. A crash between handler success and cache commit leaves the slot empty; the next retry re-executes the handler. Idempotent handlers absorb this without harm. Handlers with non-idempotent side effects (e.g., INSERT INTO media_buys without a unique constraint on the buyer's idempotency_key) need either: (a) handler-level dedupe via a database unique constraint that maps to the same key the SDK uses, or (b) the co-tx variant once it ships. Co-tx — handler passes its own connection so the cache write commits atomically with side effects — is planned as a follow-on enhancement.

Schema bootstrap caveat. :meth:create_schema uses CREATE TABLE IF NOT EXISTS — if a table with the same name but a different shape already exists (Alembic migration drift, manual DDL with response JSON instead of JSONB, missing COLLATE "C"), this method is a no-op and the backend will run against the wrong column types. If you manage the schema with Alembic / dbmate, copy the DDL inside :meth:create_schema verbatim into a migration revision — keep COLLATE "C" and JSONB identical — and skip calling :meth:create_schema at boot.

Response payload contract. :attr:CachedResponse.response is serialized via json.dumps for the JSONB column. Values must be JSON-safe — no datetime, Decimal, set, or bytes. Coerce in your handler before returning.

Cardinality / DoS. This backend has no row cap; only TTL bounds the table size. Per AdCP spec, per-principal rate limiting at the auth tier is required — the backend trusts that. Schedule :meth:delete_expired as a cron / pg_cron / app-loop sweep (get self-filters expired rows, but they accumulate on disk until something deletes them).

Schema. Created idempotently by :meth:create_schema:

.. code-block:: sql

CREATE TABLE IF NOT EXISTS adcp_idempotency (
    scope_key    TEXT        COLLATE "C" NOT NULL,
    key          TEXT        COLLATE "C" NOT NULL,
    payload_hash TEXT        NOT NULL,
    response     JSONB       NOT NULL,
    expires_at   TIMESTAMPTZ NOT NULL,
    PRIMARY KEY (scope_key, key)
);
CREATE INDEX IF NOT EXISTS adcp_idempotency_expires_idx
    ON adcp_idempotency (expires_at);

COLLATE "C" on identifier columns avoids locale-driven equivalence (Principal-A ≡ principal-a under Turkish/locale-aware collations) collapsing distinct tenants into the same cache slot.

:param pool: psycopg_pool.AsyncConnectionPool owned by the caller. Each operation acquires a short-lived connection. We don't open, own, or close the pool. :param table_name: Override the default table name. Useful for multi-tenant schema scoping. Default adcp_idempotency.

:raises ImportError: when psycopg/psycopg-pool are not installed. Install via the pg extra: pip install 'adcp[pg]'. :raises ValueError: when table_name is not a safe ASCII identifier ([a-z_][a-z0-9_]{0,62}).

Ancestors

• IdempotencyBackend
• abc.ABC

Methods

async def create_schema(self) ‑> None

Expand source code

async def create_schema(self) -> None:
    """Bootstrap the table + index. Idempotent.

    Safe to call on every app boot. Each DDL statement is executed
    separately — psycopg does not split on ``;``.
    """
    t = self._table
    statements = [
        f"""CREATE TABLE IF NOT EXISTS {t} (
            scope_key    TEXT        COLLATE "C" NOT NULL,
            key          TEXT        COLLATE "C" NOT NULL,
            payload_hash TEXT        NOT NULL,
            response     JSONB       NOT NULL,
            expires_at   TIMESTAMPTZ NOT NULL,
            PRIMARY KEY (scope_key, key)
        )""",
        # Partial-free expiry index — cheap eviction sweep.
        f"""CREATE INDEX IF NOT EXISTS {t}_expires_idx
            ON {t} (expires_at)""",
    ]
    async with self._pool.connection() as conn:
        for stmt in statements:
            await conn.execute(stmt)

Bootstrap the table + index. Idempotent.

Safe to call on every app boot. Each DDL statement is executed separately — psycopg does not split on ;.

async def delete_expired(self, now_epoch: float | None = None) ‑> int

Expand source code

async def delete_expired(self, now_epoch: float | None = None) -> int:
    """Best-effort sweep of expired entries. Returns rows removed."""
    cutoff = now_epoch if now_epoch is not None else time.time()
    cutoff_dt = datetime.fromtimestamp(cutoff, tz=timezone.utc)
    async with self._pool.connection() as conn:
        cur = await conn.execute(self._sql_delete_expired, (cutoff_dt,))
        return cur.rowcount or 0

Best-effort sweep of expired entries. Returns rows removed.

async def get(self, scope_key: str, key: str) ‑> CachedResponse | None

Expand source code

async def get(self, scope_key: str, key: str) -> CachedResponse | None:
    """Read the cached entry, filtering expired rows in the WHERE clause.

    Lazy expiry — expired rows stay on disk until ``delete_expired``
    sweeps them. ``get`` self-filters via ``expires_at > now()`` so a
    stale row never replays.
    """
    async with self._pool.connection() as conn:
        cur = await conn.execute(self._sql_get, (scope_key, key))
        row = await cur.fetchone()
        if row is None:
            return None
        payload_hash, response, expires_at = row
        return CachedResponse(
            payload_hash=payload_hash,
            response=response if isinstance(response, dict) else json.loads(response),
            expires_at_epoch=_to_epoch(expires_at),
        )

Read the cached entry, filtering expired rows in the WHERE clause.

Lazy expiry — expired rows stay on disk until delete_expired sweeps them. get self-filters via expires_at > now() so a stale row never replays.

async def put(self,
scope_key: str,
key: str,
entry: CachedResponse) ‑> None

Expand source code

async def put(
    self,
    scope_key: str,
    key: str,
    entry: CachedResponse,
) -> None:
    """Atomic upsert under ``(scope_key, key)``.

    ``ON CONFLICT DO UPDATE`` because the store only calls ``put``
    after verifying the slot is empty or expired — an overwrite in
    that window is a legitimate retry of the write itself.
    """
    expires_at_dt = datetime.fromtimestamp(entry.expires_at_epoch, tz=timezone.utc)
    async with self._pool.connection() as conn:
        await conn.execute(
            self._sql_put,
            (
                scope_key,
                key,
                entry.payload_hash,
                json.dumps(entry.response),
                expires_at_dt,
            ),
        )

Atomic upsert under (scope_key, key).

ON CONFLICT DO UPDATE because the store only calls put after verifying the slot is empty or expired — an overwrite in that window is a legitimate retry of the write itself.

class WebhookDedupStore (backend: IdempotencyBackend,
ttl_seconds: int = 86400,
*,
namespace: str = 'webhook',
clock: Callable[[], float] = <built-in function time>)

Expand source code

class WebhookDedupStore:
    """Dedup ``(sender_id, idempotency_key)`` pairs to suppress retried webhooks.

    :param backend: any :class:`IdempotencyBackend`. Same MemoryBackend or
        PgBackend type used by :class:`IdempotencyStore` is fine — the
        ``namespace`` parameter prefixes all sender IDs so request-side and
        webhook-side scopes can't alias even when sharing one backend instance.
    :param ttl_seconds: replay window. Must be within ``[86400, 604800]`` per
        the spec minimum. Defaults to 86400 (24h).
    :param namespace: prefix applied to every ``sender_id`` before it hits
        the backend. Defaults to ``"webhook"``, which is safe when the same
        backend is shared with :class:`IdempotencyStore` (request-side keys
        are scoped by a principal_id that isn't wrapped in this namespace,
        so collisions are impossible). Override only if you run multiple
        webhook scopes against one backend (e.g., separate dedup spaces for
        task webhooks vs list-change webhooks).
    """

    def __init__(
        self,
        backend: IdempotencyBackend,
        ttl_seconds: int = _MIN_TTL_SECONDS,
        *,
        namespace: str = "webhook",
        clock: Callable[[], float] = time.time,
    ) -> None:
        if not _MIN_TTL_SECONDS <= ttl_seconds <= _MAX_TTL_SECONDS:
            raise ValueError(
                f"ttl_seconds must be in [{_MIN_TTL_SECONDS}, {_MAX_TTL_SECONDS}] "
                f"per webhook spec minimum, got {ttl_seconds}"
            )
        if not namespace:
            raise ValueError("namespace must be a non-empty string")
        self.backend = backend
        self.ttl_seconds = ttl_seconds
        self.namespace = namespace
        self._clock = clock

    async def check_and_record(self, sender_id: str, idempotency_key: str) -> bool:
        """Atomically check for first-seen and record if new.

        Returns ``True`` when the pair is first-seen (event should be
        processed), ``False`` on duplicate (caller MUST still return 2xx to
        the sender — the event was delivered successfully, it's just a retry).

        Race note: the check-then-put pattern is not atomic across concurrent
        callers unless the backend provides its own atomicity. MemoryBackend
        serializes individual ``get`` and ``put`` under an ``asyncio.Lock`` but
        does NOT bracket them together — two concurrent retries of the same
        event CAN both observe "first-seen" and both process the event. That's
        a tolerable failure mode: the ultimate guarantee is "at most once per
        replay window in the common case"; a concurrent retry arriving in the
        same few milliseconds is rare and, if it happens, produces the same
        "duplicated side effect" outcome the at-least-once contract already
        warns callers to tolerate. PgBackend implementations SHOULD use
        ``INSERT ... ON CONFLICT DO NOTHING`` returning ``rowcount`` for
        lock-free atomicity.
        """
        if not sender_id:
            raise ValueError("sender_id must be a non-empty string")
        if not idempotency_key:
            raise ValueError("idempotency_key must be a non-empty string")

        scoped_sender = f"{self.namespace}:{sender_id}"
        existing = await self.backend.get(scoped_sender, idempotency_key)
        if existing is not None:
            logger.debug(
                "webhook dedup: duplicate sender=%s key_prefix=%s",
                sender_id,
                idempotency_key[:8],
            )
            return False

        entry = CachedResponse(
            payload_hash=_SENTINEL_HASH,
            response={},
            expires_at_epoch=self._clock() + self.ttl_seconds,
        )
        try:
            await self.backend.put(scoped_sender, idempotency_key, entry)
        except Exception:
            # Same fail-open reasoning as the request-side store: log and
            # process. Swallowing the put failure means this event MIGHT
            # reprocess on retry, not that we drop it. Better than raising,
            # which would look like handler failure to the sender.
            logger.warning(
                "webhook dedup put failed for sender=%s key_prefix=%s — "
                "event processed but next retry will reprocess",
                sender_id,
                idempotency_key[:8],
                exc_info=True,
            )
        return True

Dedup (sender_id, idempotency_key) pairs to suppress retried webhooks.

:param backend: any :class:IdempotencyBackend. Same MemoryBackend or PgBackend type used by :class:IdempotencyStore is fine — the namespace parameter prefixes all sender IDs so request-side and webhook-side scopes can't alias even when sharing one backend instance. :param ttl_seconds: replay window. Must be within [86400, 604800] per the spec minimum. Defaults to 86400 (24h). :param namespace: prefix applied to every sender_id before it hits the backend. Defaults to "webhook", which is safe when the same backend is shared with :class:IdempotencyStore (request-side keys are scoped by a principal_id that isn't wrapped in this namespace, so collisions are impossible). Override only if you run multiple webhook scopes against one backend (e.g., separate dedup spaces for task webhooks vs list-change webhooks).

Methods

async def check_and_record(self, sender_id: str, idempotency_key: str) ‑> bool

Expand source code

async def check_and_record(self, sender_id: str, idempotency_key: str) -> bool:
    """Atomically check for first-seen and record if new.

    Returns ``True`` when the pair is first-seen (event should be
    processed), ``False`` on duplicate (caller MUST still return 2xx to
    the sender — the event was delivered successfully, it's just a retry).

    Race note: the check-then-put pattern is not atomic across concurrent
    callers unless the backend provides its own atomicity. MemoryBackend
    serializes individual ``get`` and ``put`` under an ``asyncio.Lock`` but
    does NOT bracket them together — two concurrent retries of the same
    event CAN both observe "first-seen" and both process the event. That's
    a tolerable failure mode: the ultimate guarantee is "at most once per
    replay window in the common case"; a concurrent retry arriving in the
    same few milliseconds is rare and, if it happens, produces the same
    "duplicated side effect" outcome the at-least-once contract already
    warns callers to tolerate. PgBackend implementations SHOULD use
    ``INSERT ... ON CONFLICT DO NOTHING`` returning ``rowcount`` for
    lock-free atomicity.
    """
    if not sender_id:
        raise ValueError("sender_id must be a non-empty string")
    if not idempotency_key:
        raise ValueError("idempotency_key must be a non-empty string")

    scoped_sender = f"{self.namespace}:{sender_id}"
    existing = await self.backend.get(scoped_sender, idempotency_key)
    if existing is not None:
        logger.debug(
            "webhook dedup: duplicate sender=%s key_prefix=%s",
            sender_id,
            idempotency_key[:8],
        )
        return False

    entry = CachedResponse(
        payload_hash=_SENTINEL_HASH,
        response={},
        expires_at_epoch=self._clock() + self.ttl_seconds,
    )
    try:
        await self.backend.put(scoped_sender, idempotency_key, entry)
    except Exception:
        # Same fail-open reasoning as the request-side store: log and
        # process. Swallowing the put failure means this event MIGHT
        # reprocess on retry, not that we drop it. Better than raising,
        # which would look like handler failure to the sender.
        logger.warning(
            "webhook dedup put failed for sender=%s key_prefix=%s — "
            "event processed but next retry will reprocess",
            sender_id,
            idempotency_key[:8],
            exc_info=True,
        )
    return True

Atomically check for first-seen and record if new.

Returns True when the pair is first-seen (event should be processed), False on duplicate (caller MUST still return 2xx to the sender — the event was delivered successfully, it's just a retry).

Race note: the check-then-put pattern is not atomic across concurrent callers unless the backend provides its own atomicity. MemoryBackend serializes individual get and put under an asyncio.Lock but does NOT bracket them together — two concurrent retries of the same event CAN both observe "first-seen" and both process the event. That's a tolerable failure mode: the ultimate guarantee is "at most once per replay window in the common case"; a concurrent retry arriving in the same few milliseconds is rare and, if it happens, produces the same "duplicated side effect" outcome the at-least-once contract already warns callers to tolerate. PgBackend implementations SHOULD use INSERT … ON CONFLICT DO NOTHING returning rowcount for lock-free atomicity.