Module adcp.webhook_supervisor

Webhook delivery supervisor — retry, circuit breaker, attempt audit.

:class:~adcp.webhook_sender.WebhookSender is the transport layer: HTTP-Signatures-aware POST, one attempt, returns a result. It does not retry, does not isolate failing endpoints, and does not record attempts.

This module provides best-effort retry, circuit breaking, and attempt audit for explicit delivery calls. It is not the atomic terminal-state/outbox publisher required to advertise the AdCP 3.2 webhook retry horizon.

Two seams:

• :class:WebhookDeliverySupervisor — low-level Protocol for explicit delivery. Implementations do not become a TaskHandoff capability backend merely by satisfying this interface.

• :class:InMemoryWebhookDeliverySupervisor — development reference impl. Wraps a :class:WebhookSender with per-endpoint :class:CircuitBreaker and retry-with-jitter. Its process-local state cannot satisfy AdCP 3.2's advertised crash-durable retry-horizon contract.

Audit trail: every attempt produces a :class:DeliveryAttempt record that an optional :class:DeliveryLogSink can persist. Adopters with an existing webhook_delivery_log table wire a sink that maps the record to their schema; adopters without one can ignore the seam.

Sequence numbers: the supervisor maintains a per-sequence_key monotonic counter so adopters that need it (e.g., delivery report webhooks where the buyer correlates by sequence #) can request one via :meth:InMemoryWebhookDeliverySupervisor.next_sequence(). Sellers who need durable cross-restart sequence numbers implement the Protocol against their persistence layer.

Functions

def supervisor_or_sender(*,
sender: WebhookSender | None,
supervisor: WebhookDeliverySupervisor | None) ‑> WebhookDeliverySupervisor | WebhookSender | None

Expand source code

def supervisor_or_sender(
    *,
    sender: WebhookSender | None,
    supervisor: WebhookDeliverySupervisor | None,
) -> WebhookDeliverySupervisor | WebhookSender | None:
    """Resolve the task-webhook delivery target.

    Supervisor takes precedence when both are passed. Either alone is
    valid; both ``None`` disable SDK-managed delivery. Used by
    :mod:`adcp.decisioning.webhook_emit` to keep the call site
    polymorphic over the two surfaces (both expose ``send_mcp`` with
    compatible signatures).
    """
    return supervisor or sender

Resolve the task-webhook delivery target.

Supervisor takes precedence when both are passed. Either alone is valid; both None disable SDK-managed delivery. Used by :mod:adcp.decisioning.webhook_emit to keep the call site polymorphic over the two surfaces (both expose send_mcp with compatible signatures).

Classes

class CircuitBreakerPolicy (failure_threshold: int = 5,
success_threshold: int = 2,
open_timeout_seconds: float = 60.0)

Expand source code

@dataclass(frozen=True)
class CircuitBreakerPolicy:
    """Per-endpoint circuit breaker tuning.

    Defaults match the salesagent reference (5 consecutive failures
    open, 60s recovery test, 2 successes close).
    """

    failure_threshold: int = 5
    success_threshold: int = 2
    open_timeout_seconds: float = 60.0

Per-endpoint circuit breaker tuning.

Defaults match the salesagent reference (5 consecutive failures open, 60s recovery test, 2 successes close).

Instance variables

var failure_threshold : int

var open_timeout_seconds : float

var success_threshold : int

class CircuitState (*args, **kwds)

Expand source code

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access::

>>> Color.RED
<Color.RED: 1>

• value lookup:

>>> Color(1)
<Color.RED: 1>

• name lookup:

>>> Color['RED']
<Color.RED: 1>

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

• enum.Enum

Class variables

var CLOSED

var HALF_OPEN

var OPEN

class DeliveryAttempt (url: str,
sequence_key: str | None,
sequence_number: int | None,
attempt_number: int,
max_attempts: int,
outcome: AttemptOutcome,
http_status_code: int | None,
error_message: str | None,
response_time_ms: int,
occurred_at: datetime,
will_retry: bool,
next_retry_at: datetime | None,
task_type: str | None,
task_id: str | None,
payload_size_bytes: int | None,
notification_type: str | None = None)

Expand source code

@dataclass(frozen=True)
class DeliveryAttempt:
    """One record passed to :class:`DeliveryLogSink` per delivery attempt.

    Sellers wire a sink that maps these to their existing
    ``webhook_delivery_log`` schema. The supervisor itself doesn't
    persist anything — the sink is the BYO storage seam.

    .. note::
        ``error_message`` may contain bytes from the buyer's webhook
        receiver (sliced via ``repr()`` to 200 chars). Adopters
        persisting this field to wide-readable logs SHOULD redact
        further at the sink layer; an attacker buyer could include
        prejudicial content in their error response body.

    ``notification_type`` is provided as a passthrough for delivery
    reports (per ``schemas/cache/media-buy/get-media-buy-delivery-response.json``:
    ``scheduled`` / ``final`` / ``adjusted`` / ``delayed`` /
    ``window_update``). Task-status notifications leave this ``None``;
    adopters firing delivery reports manually populate it.
    """

    url: str
    sequence_key: str | None
    sequence_number: int | None
    attempt_number: int
    max_attempts: int
    outcome: AttemptOutcome
    http_status_code: int | None
    error_message: str | None
    response_time_ms: int
    occurred_at: datetime
    will_retry: bool
    next_retry_at: datetime | None
    task_type: str | None
    task_id: str | None
    payload_size_bytes: int | None
    notification_type: str | None = None

One record passed to :class:DeliveryLogSink per delivery attempt.

Sellers wire a sink that maps these to their existing webhook_delivery_log schema. The supervisor itself doesn't persist anything — the sink is the BYO storage seam.

Note

error_message may contain bytes from the buyer's webhook receiver (sliced via repr() to 200 chars). Adopters persisting this field to wide-readable logs SHOULD redact further at the sink layer; an attacker buyer could include prejudicial content in their error response body.

notification_type is provided as a passthrough for delivery reports (per schemas/cache/media-buy/get-media-buy-delivery-response.json: scheduled / final / adjusted / delayed / window_update). Task-status notifications leave this None; adopters firing delivery reports manually populate it.

Instance variables

var attempt_number : int

var error_message : str | None

var http_status_code : int | None

var max_attempts : int

var next_retry_at : datetime.datetime | None

var notification_type : str | None

var occurred_at : datetime.datetime

var outcome : Literal['success', 'failure', 'circuit_open']

var payload_size_bytes : int | None

var response_time_ms : int

var sequence_key : str | None

var sequence_number : int | None

var task_id : str | None

var task_type : str | None

var url : str

var will_retry : bool

class DeliveryLogSink (*args, **kwargs)

Expand source code

@runtime_checkable
class DeliveryLogSink(Protocol):
    """Optional persistence hook for delivery attempts.

    Called once per attempt. Failures inside ``record`` are
    logged-and-swallowed by the supervisor — a broken sink must not
    cascade into webhook delivery loss.

    Adopters typically wire this to a ``webhook_delivery_log`` table in
    their existing schema. No-op implementations are valid for adopters
    who don't need attempt-level audit.
    """

    async def record(self, attempt: DeliveryAttempt) -> None: ...

Optional persistence hook for delivery attempts.

Called once per attempt. Failures inside record are logged-and-swallowed by the supervisor — a broken sink must not cascade into webhook delivery loss.

Adopters typically wire this to a webhook_delivery_log table in their existing schema. No-op implementations are valid for adopters who don't need attempt-level audit.

Ancestors

• typing.Protocol
• typing.Generic

Methods

async def record(self,
attempt: DeliveryAttempt) ‑> None

Expand source code

async def record(self, attempt: DeliveryAttempt) -> None: ...

class InMemoryWebhookDeliverySupervisor (sender: WebhookSender | None,
*,
retry: RetryPolicy | None = None,
circuit: CircuitBreakerPolicy | None = None,
log_sink: DeliveryLogSink | None = None)

Expand source code

class InMemoryWebhookDeliverySupervisor:
    """Reference :class:`WebhookDeliverySupervisor` with in-process state.

    Development-only for AdCP 3.2 webhook publishers: circuit breaker and
    retry state live in the running process, so this implementation cannot
    back an advertised delivery retry horizon. Production publishers use a
    durable outbox that also retains immutable delivery-key bindings.

    Wraps an underlying :class:`~adcp.webhook_sender.WebhookSender` for
    actual HTTP-Signatures send. Per-endpoint state isolation: each
    URL gets its own breaker so one buyer's outage doesn't quarantine
    another.

    Sequence numbers: per-``sequence_key`` monotonic counters live in
    memory. Sellers that need cross-restart sequence numbers (e.g., to
    persist delivery report sequence # in their database) implement the
    Protocol directly against their persistence layer instead.
    """

    delivery_state_is_durable = False
    delivery_retry_horizon_seconds: int | None = None

    def __init__(
        self,
        sender: WebhookSender | None,
        *,
        retry: RetryPolicy | None = None,
        circuit: CircuitBreakerPolicy | None = None,
        log_sink: DeliveryLogSink | None = None,
    ) -> None:
        # Fail-fast: a None sender would later AttributeError on every
        # send_mcp call. The boot-time webhook gate also passes when a
        # supervisor is wired, so we'd defeat its fail-fast intent
        # without this check.
        if sender is None:
            raise ValueError(
                "InMemoryWebhookDeliverySupervisor requires a non-None "
                "WebhookSender. Construct one via WebhookSender.from_jwk(...) "
                "or WebhookSender.from_pem(...) and pass it as the first "
                "positional argument."
            )
        self._sender = sender
        self._retry = retry or RetryPolicy()
        self._circuit_policy = circuit or CircuitBreakerPolicy()
        self._log_sink = log_sink
        # Breakers keyed on (breaker_key or url). When two tenants share
        # a SaaS receiver URL (Zapier, Make, etc.), pass a tenant-scoped
        # ``breaker_key`` to ``send_mcp`` so one tenant's failures don't
        # quarantine another tenant's deliveries.
        self._breakers: dict[str, _CircuitBreaker] = {}
        # Per-``sequence_key`` monotonic counter. Recommend using a
        # per-stream key like ``f"{media_buy_id}:{url}"`` for delivery
        # reports — multiple subscribers per media_buy each need their
        # own sequence numbering. In-memory only; multi-instance
        # deployments will collide on the wire and need to implement
        # the Protocol against durable storage. This warning fires once
        # at first allocation so operators see it on cold start.
        self._sequence_numbers: dict[str, int] = {}
        self._sequence_warned = False
        self._state_lock = threading.Lock()

    def next_sequence(self, key: str) -> int:
        """Allocate the next sequence number for ``key`` (1-indexed).

        Recommend a per-stream key like ``f"{media_buy_id}:{url}"`` for
        delivery-report webhooks — multiple subscribers per media buy
        each need their own monotonic stream. ``media_buy_id`` alone
        would have all subscribers share a sequence, which the spec's
        delivery-report semantics don't support (each receiver sees
        non-contiguous numbers).

        Sellers who manage sequence numbers themselves (e.g., persisted
        in a database column for cross-restart durability) ignore this
        and pass their own value via wherever it's used downstream.
        """
        with self._state_lock:
            if not self._sequence_warned:
                self._sequence_warned = True
                logger.info(
                    "[adcp.webhook_supervisor] in-process sequence numbers "
                    "are not durable across process restart and not shared "
                    "across multi-worker deployments. Sellers needing "
                    "cross-restart sequence semantics should implement "
                    "WebhookDeliverySupervisor against their persistence "
                    "layer."
                )
            current = self._sequence_numbers.get(key, 0) + 1
            self._sequence_numbers[key] = current
            return current

    def _breaker_for(self, key: str) -> _CircuitBreaker:
        with self._state_lock:
            breaker = self._breakers.get(key)
            if breaker is None:
                breaker = _CircuitBreaker(self._circuit_policy)
                self._breakers[key] = breaker
            return breaker

    async def send_mcp(
        self,
        *,
        url: str,
        task_id: str,
        status: GeneratedTaskStatus | str,
        task_type: TaskType | str,
        result: Any = None,
        operation_id: str,
        token: str | None = None,
        sequence_key: str | None = None,
        breaker_key: str | None = None,
        notification_type: str | None = None,
    ) -> WebhookDeliveryResult | None:
        """Deliver one MCP-style webhook with retry + circuit-breaker.

        Returns the final attempt's :class:`WebhookDeliveryResult` on
        success or the last failed attempt; returns ``None`` when the
        breaker is OPEN and no attempt is made (the audit log records
        a ``circuit_open`` row regardless).

        :param breaker_key: Override the circuit-breaker key (default:
            ``url``). Multi-tenant sellers whose buyers register
            shared SaaS receiver URLs (Zapier, Make, etc.) MUST pass a
            tenant-scoped key (e.g., ``f"{tenant_id}:{url}"``) so one
            tenant's failures don't quarantine deliveries to the same
            URL for another tenant.
        :param sequence_key: Allocates a per-stream sequence number
            via :meth:`next_sequence`. Recommend
            ``f"{media_buy_id}:{url}"`` (per-receiver stream); see
            :meth:`next_sequence` for the rationale.
        :param operation_id: Buyer-supplied correlation id from
            ``push_notification_config.operation_id``; echoed verbatim
            into the webhook payload's ``operation_id`` field per spec.
            Threaded through to the underlying
            :meth:`WebhookSender.send_mcp`. On retry (``resend``) the id
            is preserved by replaying the exact attempt-1 bytes.
        :param notification_type: Passthrough to ``DeliveryAttempt``
            for delivery-report webhooks (``scheduled`` / ``final`` /
            ``adjusted`` / ``delayed`` / ``window_update``).

        **Idempotency-key reuse on retry** (per spec
        ``mcp-webhook-payload.json``: "Publishers MUST … reuse the
        same key on every retry"): attempts 2+ replay the exact bytes
        of attempt 1 via :meth:`WebhookSender.resend`, preserving the
        ``idempotency_key`` for receiver-side dedup. Only attempt 1
        (or any attempt whose predecessor raised before producing a
        result) calls ``send_mcp`` fresh.

        Each attempt is logged to the :class:`DeliveryLogSink` if one
        is configured. Sink failures are swallowed; a slow sink is
        bounded by ``RetryPolicy.sink_timeout_seconds``.

        Each attempt is logged to the :class:`DeliveryLogSink` if one
        is configured. Sink failures are swallowed.
        """
        breaker = self._breaker_for(breaker_key or url)
        sequence_number: int | None = None  # allocated AFTER breaker check
        # Audit log + DeliveryAttempt store the on-wire string; TaskType
        # enum is normalized once here so every record sees the same value.
        task_type_str: str = task_type.value if isinstance(task_type, TaskType) else task_type

        if not breaker.can_attempt():
            await self._record(
                DeliveryAttempt(
                    url=url,
                    sequence_key=sequence_key,
                    sequence_number=None,
                    attempt_number=0,
                    max_attempts=self._retry.max_attempts,
                    outcome="circuit_open",
                    http_status_code=None,
                    error_message=(f"circuit breaker OPEN for {url} — skipped delivery"),
                    response_time_ms=0,
                    occurred_at=datetime.now(UTC),
                    will_retry=False,
                    next_retry_at=None,
                    task_type=task_type_str,
                    task_id=task_id,
                    payload_size_bytes=None,
                    notification_type=notification_type,
                )
            )
            logger.warning(
                "[adcp.webhook_supervisor] circuit OPEN for %s — skipped %s",
                url,
                task_type or "webhook",
            )
            return None

        # Allocate sequence number AFTER the breaker check so we don't
        # burn numbers on circuit-open skips. Once allocated, the same
        # sequence number is used for every attempt of THIS delivery
        # (per-delivery, not per-attempt).
        if sequence_key is not None:
            sequence_number = self.next_sequence(sequence_key)

        last_result: WebhookDeliveryResult | None = None
        for attempt_number in range(1, self._retry.max_attempts + 1):
            delay = self._retry.delay_for_attempt(attempt_number)
            if delay > 0:
                await asyncio.sleep(delay)

            attempt_started = datetime.now(UTC)
            attempt_started_monotonic = time.monotonic()
            try:
                # Spec-compliant retry: replay the exact bytes (same
                # idempotency_key) on attempts 2+ via ``resend``. Only
                # attempt 1, or attempts whose predecessor raised
                # before producing a result with bytes, call send_mcp
                # fresh.
                if last_result is not None and last_result.sent_body:
                    last_result = await self._sender.resend(last_result)
                else:
                    last_result = await self._sender.send_mcp(
                        url=url,
                        task_id=task_id,
                        status=status,
                        task_type=task_type,
                        result=result,
                        operation_id=operation_id,
                        token=token,
                    )
                response_time_ms = int((time.monotonic() - attempt_started_monotonic) * 1000)
                if last_result.ok:
                    breaker.record_success()
                    await self._record(
                        DeliveryAttempt(
                            url=url,
                            sequence_key=sequence_key,
                            sequence_number=sequence_number,
                            attempt_number=attempt_number,
                            max_attempts=self._retry.max_attempts,
                            outcome="success",
                            http_status_code=last_result.status_code,
                            error_message=None,
                            response_time_ms=response_time_ms,
                            occurred_at=attempt_started,
                            will_retry=False,
                            next_retry_at=None,
                            task_type=task_type_str,
                            task_id=task_id,
                            payload_size_bytes=len(last_result.sent_body),
                            notification_type=notification_type,
                        )
                    )
                    return last_result
                # Non-2xx — count as failure for breaker, retry
                will_retry = attempt_number < self._retry.max_attempts
                next_delay = (
                    self._retry.delay_for_attempt(attempt_number + 1) if will_retry else None
                )
                next_retry_at = (
                    attempt_started + timedelta(seconds=next_delay)
                    if next_delay is not None
                    else None
                )
                breaker.record_failure()
                await self._record(
                    DeliveryAttempt(
                        url=url,
                        sequence_key=sequence_key,
                        sequence_number=sequence_number,
                        attempt_number=attempt_number,
                        max_attempts=self._retry.max_attempts,
                        outcome="failure",
                        http_status_code=last_result.status_code,
                        error_message=(
                            f"HTTP {last_result.status_code}: "
                            f"{last_result.response_body[:200]!r}"
                        ),
                        response_time_ms=response_time_ms,
                        occurred_at=attempt_started,
                        will_retry=will_retry,
                        next_retry_at=next_retry_at,
                        task_type=task_type_str,
                        task_id=task_id,
                        payload_size_bytes=len(last_result.sent_body),
                        notification_type=notification_type,
                    )
                )
            except Exception as exc:
                response_time_ms = int((time.monotonic() - attempt_started_monotonic) * 1000)
                will_retry = attempt_number < self._retry.max_attempts
                next_delay = (
                    self._retry.delay_for_attempt(attempt_number + 1) if will_retry else None
                )
                next_retry_at = (
                    attempt_started + timedelta(seconds=next_delay)
                    if next_delay is not None
                    else None
                )
                breaker.record_failure()
                await self._record(
                    DeliveryAttempt(
                        url=url,
                        sequence_key=sequence_key,
                        sequence_number=sequence_number,
                        attempt_number=attempt_number,
                        max_attempts=self._retry.max_attempts,
                        outcome="failure",
                        http_status_code=None,
                        error_message=f"{type(exc).__name__}: {exc}",
                        response_time_ms=response_time_ms,
                        occurred_at=attempt_started,
                        will_retry=will_retry,
                        next_retry_at=next_retry_at,
                        task_type=task_type_str,
                        task_id=task_id,
                        payload_size_bytes=None,
                        notification_type=notification_type,
                    )
                )
                if not will_retry:
                    raise

        return last_result

    async def _record(self, attempt: DeliveryAttempt) -> None:
        """Persist one attempt to the configured sink.

        Bounded by ``RetryPolicy.sink_timeout_seconds`` (default 5s)
        so a misbehaving sink (DB stall, lock contention, unbounded
        queue) cannot freeze the supervisor's hot path. Both timeout
        and exception are logged-and-swallowed — a broken sink must
        not cascade into webhook delivery loss.
        """
        if self._log_sink is None:
            return
        try:
            await asyncio.wait_for(
                self._log_sink.record(attempt),
                timeout=self._retry.sink_timeout_seconds,
            )
        except asyncio.TimeoutError:
            logger.warning(
                "[adcp.webhook_supervisor] DeliveryLogSink timed out "
                "after %ss on attempt for %s — log dropped, delivery "
                "unaffected",
                self._retry.sink_timeout_seconds,
                attempt.url,
            )
        except Exception:
            logger.warning(
                "[adcp.webhook_supervisor] DeliveryLogSink raised on attempt "
                "for %s — log dropped, delivery unaffected",
                attempt.url,
                exc_info=True,
            )

Reference :class:WebhookDeliverySupervisor with in-process state.

Development-only for AdCP 3.2 webhook publishers: circuit breaker and retry state live in the running process, so this implementation cannot back an advertised delivery retry horizon. Production publishers use a durable outbox that also retains immutable delivery-key bindings.

Wraps an underlying :class:~adcp.webhook_sender.WebhookSender for actual HTTP-Signatures send. Per-endpoint state isolation: each URL gets its own breaker so one buyer's outage doesn't quarantine another.

Sequence numbers: per-sequence_key monotonic counters live in memory. Sellers that need cross-restart sequence numbers (e.g., to persist delivery report sequence # in their database) implement the Protocol directly against their persistence layer instead.

Class variables

var delivery_retry_horizon_seconds : int | None

var delivery_state_is_durable

Methods

def next_sequence(self, key: str) ‑> int

Expand source code

def next_sequence(self, key: str) -> int:
    """Allocate the next sequence number for ``key`` (1-indexed).

    Recommend a per-stream key like ``f"{media_buy_id}:{url}"`` for
    delivery-report webhooks — multiple subscribers per media buy
    each need their own monotonic stream. ``media_buy_id`` alone
    would have all subscribers share a sequence, which the spec's
    delivery-report semantics don't support (each receiver sees
    non-contiguous numbers).

    Sellers who manage sequence numbers themselves (e.g., persisted
    in a database column for cross-restart durability) ignore this
    and pass their own value via wherever it's used downstream.
    """
    with self._state_lock:
        if not self._sequence_warned:
            self._sequence_warned = True
            logger.info(
                "[adcp.webhook_supervisor] in-process sequence numbers "
                "are not durable across process restart and not shared "
                "across multi-worker deployments. Sellers needing "
                "cross-restart sequence semantics should implement "
                "WebhookDeliverySupervisor against their persistence "
                "layer."
            )
        current = self._sequence_numbers.get(key, 0) + 1
        self._sequence_numbers[key] = current
        return current

Allocate the next sequence number for key (1-indexed).

Recommend a per-stream key like f"{media_buy_id}:{url}" for delivery-report webhooks — multiple subscribers per media buy each need their own monotonic stream. media_buy_id alone would have all subscribers share a sequence, which the spec's delivery-report semantics don't support (each receiver sees non-contiguous numbers).

Sellers who manage sequence numbers themselves (e.g., persisted in a database column for cross-restart durability) ignore this and pass their own value via wherever it's used downstream.

async def send_mcp(self,
*,
url: str,
task_id: str,
status: GeneratedTaskStatus | str,
task_type: TaskType | str,
result: Any = None,
operation_id: str,
token: str | None = None,
sequence_key: str | None = None,
breaker_key: str | None = None,
notification_type: str | None = None) ‑> WebhookDeliveryResult | None

Expand source code

async def send_mcp(
    self,
    *,
    url: str,
    task_id: str,
    status: GeneratedTaskStatus | str,
    task_type: TaskType | str,
    result: Any = None,
    operation_id: str,
    token: str | None = None,
    sequence_key: str | None = None,
    breaker_key: str | None = None,
    notification_type: str | None = None,
) -> WebhookDeliveryResult | None:
    """Deliver one MCP-style webhook with retry + circuit-breaker.

    Returns the final attempt's :class:`WebhookDeliveryResult` on
    success or the last failed attempt; returns ``None`` when the
    breaker is OPEN and no attempt is made (the audit log records
    a ``circuit_open`` row regardless).

    :param breaker_key: Override the circuit-breaker key (default:
        ``url``). Multi-tenant sellers whose buyers register
        shared SaaS receiver URLs (Zapier, Make, etc.) MUST pass a
        tenant-scoped key (e.g., ``f"{tenant_id}:{url}"``) so one
        tenant's failures don't quarantine deliveries to the same
        URL for another tenant.
    :param sequence_key: Allocates a per-stream sequence number
        via :meth:`next_sequence`. Recommend
        ``f"{media_buy_id}:{url}"`` (per-receiver stream); see
        :meth:`next_sequence` for the rationale.
    :param operation_id: Buyer-supplied correlation id from
        ``push_notification_config.operation_id``; echoed verbatim
        into the webhook payload's ``operation_id`` field per spec.
        Threaded through to the underlying
        :meth:`WebhookSender.send_mcp`. On retry (``resend``) the id
        is preserved by replaying the exact attempt-1 bytes.
    :param notification_type: Passthrough to ``DeliveryAttempt``
        for delivery-report webhooks (``scheduled`` / ``final`` /
        ``adjusted`` / ``delayed`` / ``window_update``).

    **Idempotency-key reuse on retry** (per spec
    ``mcp-webhook-payload.json``: "Publishers MUST … reuse the
    same key on every retry"): attempts 2+ replay the exact bytes
    of attempt 1 via :meth:`WebhookSender.resend`, preserving the
    ``idempotency_key`` for receiver-side dedup. Only attempt 1
    (or any attempt whose predecessor raised before producing a
    result) calls ``send_mcp`` fresh.

    Each attempt is logged to the :class:`DeliveryLogSink` if one
    is configured. Sink failures are swallowed; a slow sink is
    bounded by ``RetryPolicy.sink_timeout_seconds``.

    Each attempt is logged to the :class:`DeliveryLogSink` if one
    is configured. Sink failures are swallowed.
    """
    breaker = self._breaker_for(breaker_key or url)
    sequence_number: int | None = None  # allocated AFTER breaker check
    # Audit log + DeliveryAttempt store the on-wire string; TaskType
    # enum is normalized once here so every record sees the same value.
    task_type_str: str = task_type.value if isinstance(task_type, TaskType) else task_type

    if not breaker.can_attempt():
        await self._record(
            DeliveryAttempt(
                url=url,
                sequence_key=sequence_key,
                sequence_number=None,
                attempt_number=0,
                max_attempts=self._retry.max_attempts,
                outcome="circuit_open",
                http_status_code=None,
                error_message=(f"circuit breaker OPEN for {url} — skipped delivery"),
                response_time_ms=0,
                occurred_at=datetime.now(UTC),
                will_retry=False,
                next_retry_at=None,
                task_type=task_type_str,
                task_id=task_id,
                payload_size_bytes=None,
                notification_type=notification_type,
            )
        )
        logger.warning(
            "[adcp.webhook_supervisor] circuit OPEN for %s — skipped %s",
            url,
            task_type or "webhook",
        )
        return None

    # Allocate sequence number AFTER the breaker check so we don't
    # burn numbers on circuit-open skips. Once allocated, the same
    # sequence number is used for every attempt of THIS delivery
    # (per-delivery, not per-attempt).
    if sequence_key is not None:
        sequence_number = self.next_sequence(sequence_key)

    last_result: WebhookDeliveryResult | None = None
    for attempt_number in range(1, self._retry.max_attempts + 1):
        delay = self._retry.delay_for_attempt(attempt_number)
        if delay > 0:
            await asyncio.sleep(delay)

        attempt_started = datetime.now(UTC)
        attempt_started_monotonic = time.monotonic()
        try:
            # Spec-compliant retry: replay the exact bytes (same
            # idempotency_key) on attempts 2+ via ``resend``. Only
            # attempt 1, or attempts whose predecessor raised
            # before producing a result with bytes, call send_mcp
            # fresh.
            if last_result is not None and last_result.sent_body:
                last_result = await self._sender.resend(last_result)
            else:
                last_result = await self._sender.send_mcp(
                    url=url,
                    task_id=task_id,
                    status=status,
                    task_type=task_type,
                    result=result,
                    operation_id=operation_id,
                    token=token,
                )
            response_time_ms = int((time.monotonic() - attempt_started_monotonic) * 1000)
            if last_result.ok:
                breaker.record_success()
                await self._record(
                    DeliveryAttempt(
                        url=url,
                        sequence_key=sequence_key,
                        sequence_number=sequence_number,
                        attempt_number=attempt_number,
                        max_attempts=self._retry.max_attempts,
                        outcome="success",
                        http_status_code=last_result.status_code,
                        error_message=None,
                        response_time_ms=response_time_ms,
                        occurred_at=attempt_started,
                        will_retry=False,
                        next_retry_at=None,
                        task_type=task_type_str,
                        task_id=task_id,
                        payload_size_bytes=len(last_result.sent_body),
                        notification_type=notification_type,
                    )
                )
                return last_result
            # Non-2xx — count as failure for breaker, retry
            will_retry = attempt_number < self._retry.max_attempts
            next_delay = (
                self._retry.delay_for_attempt(attempt_number + 1) if will_retry else None
            )
            next_retry_at = (
                attempt_started + timedelta(seconds=next_delay)
                if next_delay is not None
                else None
            )
            breaker.record_failure()
            await self._record(
                DeliveryAttempt(
                    url=url,
                    sequence_key=sequence_key,
                    sequence_number=sequence_number,
                    attempt_number=attempt_number,
                    max_attempts=self._retry.max_attempts,
                    outcome="failure",
                    http_status_code=last_result.status_code,
                    error_message=(
                        f"HTTP {last_result.status_code}: "
                        f"{last_result.response_body[:200]!r}"
                    ),
                    response_time_ms=response_time_ms,
                    occurred_at=attempt_started,
                    will_retry=will_retry,
                    next_retry_at=next_retry_at,
                    task_type=task_type_str,
                    task_id=task_id,
                    payload_size_bytes=len(last_result.sent_body),
                    notification_type=notification_type,
                )
            )
        except Exception as exc:
            response_time_ms = int((time.monotonic() - attempt_started_monotonic) * 1000)
            will_retry = attempt_number < self._retry.max_attempts
            next_delay = (
                self._retry.delay_for_attempt(attempt_number + 1) if will_retry else None
            )
            next_retry_at = (
                attempt_started + timedelta(seconds=next_delay)
                if next_delay is not None
                else None
            )
            breaker.record_failure()
            await self._record(
                DeliveryAttempt(
                    url=url,
                    sequence_key=sequence_key,
                    sequence_number=sequence_number,
                    attempt_number=attempt_number,
                    max_attempts=self._retry.max_attempts,
                    outcome="failure",
                    http_status_code=None,
                    error_message=f"{type(exc).__name__}: {exc}",
                    response_time_ms=response_time_ms,
                    occurred_at=attempt_started,
                    will_retry=will_retry,
                    next_retry_at=next_retry_at,
                    task_type=task_type_str,
                    task_id=task_id,
                    payload_size_bytes=None,
                    notification_type=notification_type,
                )
            )
            if not will_retry:
                raise

    return last_result

Deliver one MCP-style webhook with retry + circuit-breaker.

Returns the final attempt's :class:WebhookDeliveryResult on success or the last failed attempt; returns None when the breaker is OPEN and no attempt is made (the audit log records a circuit_open row regardless).

:param breaker_key: Override the circuit-breaker key (default: url). Multi-tenant sellers whose buyers register shared SaaS receiver URLs (Zapier, Make, etc.) MUST pass a tenant-scoped key (e.g., f"{tenant_id}:{url}") so one tenant's failures don't quarantine deliveries to the same URL for another tenant. :param sequence_key: Allocates a per-stream sequence number via :meth:next_sequence. Recommend f"{media_buy_id}:{url}" (per-receiver stream); see :meth:next_sequence for the rationale. :param operation_id: Buyer-supplied correlation id from push_notification_config.operation_id; echoed verbatim into the webhook payload's operation_id field per spec. Threaded through to the underlying :meth:WebhookSender.send_mcp. On retry (resend) the id is preserved by replaying the exact attempt-1 bytes. :param notification_type: Passthrough to DeliveryAttempt for delivery-report webhooks (scheduled / final / adjusted / delayed / window_update).

Idempotency-key reuse on retry (per spec mcp-webhook-payload.json: "Publishers MUST … reuse the same key on every retry"): attempts 2+ replay the exact bytes of attempt 1 via :meth:WebhookSender.resend, preserving the idempotency_key for receiver-side dedup. Only attempt 1 (or any attempt whose predecessor raised before producing a result) calls send_mcp fresh.

Each attempt is logged to the :class:DeliveryLogSink if one is configured. Sink failures are swallowed; a slow sink is bounded by RetryPolicy.sink_timeout_seconds.

Each attempt is logged to the :class:DeliveryLogSink if one is configured. Sink failures are swallowed.

class RetryPolicy (max_attempts: int = 3,
base_delay_seconds: float = 1.0,
max_delay_seconds: float = 30.0,
jitter: bool = True,
sink_timeout_seconds: float = 5.0)

Expand source code

@dataclass(frozen=True)
class RetryPolicy:
    """Retry behavior for a single :meth:`send_mcp` call.

    Defaults match the salesagent reference adopter (3 attempts,
    exponential backoff with jitter, 1s base, 30s cap). Adopters with
    different SLAs override per-supervisor.

    ``sink_timeout_seconds`` bounds how long a slow
    :class:`DeliveryLogSink.record` call can stall the supervisor's
    hot path (default 5s). Sink timeouts log a warning and continue
    delivery — a misbehaving sink must not be able to wedge the
    delivery pipeline.
    """

    max_attempts: int = 3
    base_delay_seconds: float = 1.0
    max_delay_seconds: float = 30.0
    jitter: bool = True
    sink_timeout_seconds: float = 5.0

    def delay_for_attempt(self, attempt_number: int) -> float:
        """Compute the sleep before ``attempt_number`` (1-indexed).

        Attempt 1 has no preceding delay; attempts 2+ use exponential
        backoff capped at ``max_delay_seconds`` with optional jitter.
        """
        if attempt_number <= 1:
            return 0.0
        delay: float = min(
            self.base_delay_seconds * (2 ** (attempt_number - 2)),
            self.max_delay_seconds,
        )
        if self.jitter:
            delay = delay * (0.5 + random.random() * 0.5)
        return delay

Retry behavior for a single :meth:send_mcp call.

Defaults match the salesagent reference adopter (3 attempts, exponential backoff with jitter, 1s base, 30s cap). Adopters with different SLAs override per-supervisor.

sink_timeout_seconds bounds how long a slow :class:DeliveryLogSink.record() call can stall the supervisor's hot path (default 5s). Sink timeouts log a warning and continue delivery — a misbehaving sink must not be able to wedge the delivery pipeline.

Instance variables

var base_delay_seconds : float

var jitter : bool

var max_attempts : int

var max_delay_seconds : float

var sink_timeout_seconds : float

Methods

def delay_for_attempt(self, attempt_number: int) ‑> float

Expand source code

def delay_for_attempt(self, attempt_number: int) -> float:
    """Compute the sleep before ``attempt_number`` (1-indexed).

    Attempt 1 has no preceding delay; attempts 2+ use exponential
    backoff capped at ``max_delay_seconds`` with optional jitter.
    """
    if attempt_number <= 1:
        return 0.0
    delay: float = min(
        self.base_delay_seconds * (2 ** (attempt_number - 2)),
        self.max_delay_seconds,
    )
    if self.jitter:
        delay = delay * (0.5 + random.random() * 0.5)
    return delay

Compute the sleep before attempt_number (1-indexed).

Attempt 1 has no preceding delay; attempts 2+ use exponential backoff capped at max_delay_seconds with optional jitter.

class WebhookDeliverySupervisor (*args, **kwargs)

Expand source code

@runtime_checkable
class WebhookDeliverySupervisor(Protocol):
    """Reliable webhook delivery surface.

    Conforms to the ``send_mcp`` shape used by asynchronous task webhooks.
    This is a low-level/best-effort transport abstraction. It is not the
    atomic task-state/outbox contract required for SDK-managed AdCP 3.2
    TaskHandoff publication; adopters advertising that capability must own
    publication and reconciliation outside this SDK.

    The reference :class:`InMemoryWebhookDeliverySupervisor` wraps a
    :class:`~adcp.webhook_sender.WebhookSender` with circuit breaker +
    retry; sellers with their own infra implement the Protocol
    directly without using the reference.
    """

    delivery_state_is_durable: bool
    delivery_retry_horizon_seconds: int | None

    async def send_mcp(
        self,
        *,
        url: str,
        task_id: str,
        status: GeneratedTaskStatus | str,
        task_type: TaskType | str,
        result: Any = None,
        operation_id: str,
        token: str | None = None,
        sequence_key: str | None = None,
        breaker_key: str | None = None,
        notification_type: str | None = None,
    ) -> WebhookDeliveryResult | None: ...

Reliable webhook delivery surface.

Conforms to the send_mcp shape used by asynchronous task webhooks. This is a low-level/best-effort transport abstraction. It is not the atomic task-state/outbox contract required for SDK-managed AdCP 3.2 TaskHandoff publication; adopters advertising that capability must own publication and reconciliation outside this SDK.

The reference :class:InMemoryWebhookDeliverySupervisor wraps a :class:~adcp.webhook_sender.WebhookSender with circuit breaker + retry; sellers with their own infra implement the Protocol directly without using the reference.

Ancestors

• typing.Protocol
• typing.Generic

Class variables

var delivery_retry_horizon_seconds : int | None

var delivery_state_is_durable : bool

Methods

async def send_mcp(self,
*,
url: str,
task_id: str,
status: GeneratedTaskStatus | str,
task_type: TaskType | str,
result: Any = None,
operation_id: str,
token: str | None = None,
sequence_key: str | None = None,
breaker_key: str | None = None,
notification_type: str | None = None) ‑> WebhookDeliveryResult | None

Expand source code

async def send_mcp(
    self,
    *,
    url: str,
    task_id: str,
    status: GeneratedTaskStatus | str,
    task_type: TaskType | str,
    result: Any = None,
    operation_id: str,
    token: str | None = None,
    sequence_key: str | None = None,
    breaker_key: str | None = None,
    notification_type: str | None = None,
) -> WebhookDeliveryResult | None: ...