diff --git a/MIGRATION_v7_to_v8.md b/MIGRATION_v7_to_v8.md index ad9ce31d6..cc0e56c31 100644 --- a/MIGRATION_v7_to_v8.md +++ b/MIGRATION_v7_to_v8.md @@ -182,6 +182,30 @@ Task webhook correlation remains unchanged: buyers supply `push_notification_config.operation_id`, and sellers echo that value in the corresponding `McpWebhookPayload.operation_id`. +### 8.2.0: scheduled delivery reports echo the registration's `operation_id` + +AdCP 3.3 adds an optional `reporting_webhook.operation_id` +([adcp#7895](https://github.com/adcontextprotocol/adcp/pull/7895)). SDK 8 still +has no typed field: a buyer-supplied value is kept as a `ReportingWebhook` +extra. + +- Buyers: set `operation_id` on `reporting_webhook` once the seller confirms + support. +- Sellers: send scheduled reports with `WebhookSender.send_media_buy_delivery` + (or `prepare_media_buy_delivery`, or `create_media_buy_delivery_webhook_payload` + with `deliver`). They echo the stored value and `token`, and validate the + value against the 3.3 constraints. A registration without one raises + `ReportingWebhookOperationIdMissingError` before any network I/O and logs a + structured WARNING; ask that buyer to update the registration rather than + inventing an ID. +- **Behavior change:** `create_mcp_webhook_payload` (and so `send_mcp` / + `prepare_mcp`) no longer defaults `notification_id` to `"{task_id}.terminal"` + for `task_type="media_buy_delivery"`. Delivery reports are point-in-time + events deduped by `idempotency_key`. Pass `notification_id` explicitly if you + relied on the old value. Other task types keep the default. + +See [reporting correlation](docs/webhooks/reporting-operation-id-3.2.md). + ## Webhook activity metadata `ActivityType.WEBHOOK_RECEIVED` no longer copies the complete callback into diff --git a/docs/webhooks/migration-from-fragmented-senders.md b/docs/webhooks/migration-from-fragmented-senders.md index f4e97f7ec..e8eb2eac3 100644 --- a/docs/webhooks/migration-from-fragmented-senders.md +++ b/docs/webhooks/migration-from-fragmented-senders.md @@ -290,3 +290,11 @@ if not result.ok: * **Hooks bypassing SSRF** — hooks run before SSRF, but SSRF validates the post-rewrite URL; the boundary holds. If your current sender doesn't have one of these properties, this is what you're trading up to. + +## Reporting-stream correlation on AdCP 3.2 + +For scheduled `media_buy_delivery` notifications, use +`WebhookSender.send_media_buy_delivery` with the stored `reporting_webhook` +registration. See the [reporting operation ID guide](reporting-operation-id-3.2.md), +which covers the registration field, seller echo and skip-and-flag behavior, and +updating existing registrations before scheduled delivery resumes. diff --git a/docs/webhooks/reporting-operation-id-3.2.md b/docs/webhooks/reporting-operation-id-3.2.md new file mode 100644 index 000000000..1dc00ff41 --- /dev/null +++ b/docs/webhooks/reporting-operation-id-3.2.md @@ -0,0 +1,118 @@ +# Reporting correlation on AdCP 3.2 + +Every `media_buy_delivery` webhook envelope requires `operation_id`, and the +seller must echo a value the buyer supplied. The published 3.2.x +`reporting_webhook` registration schema has no field for it, so a buyer that +registers only `reporting_webhook` had no way to supply one. + +AdCP 3.3 fixes this. [adcp#7895](https://github.com/adcontextprotocol/adcp/pull/7895) +(closing [adcp#7885](https://github.com/adcontextprotocol/adcp/issues/7885)) adds +an optional `reporting_webhook.operation_id` with the same constraints as +`push_notification_config.operation_id`: 1–255 characters from `A–Z`, `a–z`, +`0–9`, `_`, `.`, `:` and `-`. The seller MUST echo it verbatim on every +`media_buy_delivery` report to that registration and MUST NOT derive it from the +URL. A seller MUST NOT send `media_buy_delivery` reports to a registration that +omits it. The field is expected to become required in 4.0. + +SDK 8 is pinned to AdCP 3.2, whose generated `ReportingWebhook` does not +declare `operation_id`. The model allows additional fields, so a buyer-supplied +`operation_id` is kept as a model extra and serialized unchanged; no +`ext.{namespace}` wrapper is needed. SDK 8 does not add a typed field or validate +the value when the registration is built. The seller-side delivery helpers below +read the extra and validate it against the 3.3 constraints before sending. On +3.2 the field is opt-in: a 3.2 seller may accept and ignore it, so confirm +seller support during onboarding. + +## Buyers: set `operation_id` on `reporting_webhook` + +```python +from adcp.types import ReportingWebhook + +registration = ReportingWebhook( + url="https://buyer.example.com/webhooks/reporting", + operation_id="op_reporting_001", + authentication={ + "schemes": ["Bearer"], + "credentials": "example_reporting_credential_32_chars", + }, + reporting_frequency="daily", +) +``` + +Generate a unique value per reporting registration (a UUID works). It is +independent of any `push_notification_config.operation_id` on the same request, +and it is not a per-delivery `idempotency_key`. Pass the registration to +`buy_products`, `accept_proposal` or `control_media_buy`, or to the legacy +`create_media_buy` / `update_media_buy` where supported. + +Correlate received reports by the payload's `operation_id`, not by URL. If you +already registered `reporting_webhook` without an ID, update the registration +with one. A seller that follows the fail-closed rule sends you no scheduled +reports until you do. Use `get_media_buy_delivery` in the meantime. + +Omitting the field is valid. Because SDK 8 has no typed field, the registration +model does not check the value: send a 1–255 character `[A-Za-z0-9_.:-]` string, +because a seller using the SDK helpers refuses anything else. Read it back with +`registration.model_extra["operation_id"]`. Historical version-scoped models keep +their pinned schemas, including the beta.6 contract in `adcp.types.v32`, where +the field is required. + +## Sellers: echo the stored value, skip registrations without one + +Store the registration as the buyer sent it, for example with +`registration.model_dump_json()`. When a report is due, build the webhook from +that stored registration. The URL, `operation_id` and `token` then all come from +the same record: + +```python +from adcp.webhooks import ReportingWebhookOperationIdMissingError, WebhookSender + +async with WebhookSender.from_bearer_token(credentials) as sender: + for buy in due_media_buys: + try: + result = await sender.send_media_buy_delivery( + registration=buy.reporting_webhook, # stored model or JSON dict + task_id=f"delivery_report_{buy.media_buy_id}_{buy.sequence:06d}", + result=report_for(buy), # MediaBuyDeliveryWebhookResult or dict + ) + except ReportingWebhookOperationIdMissingError: + flag_for_buyer(buy) # registration has no operation_id; no request was sent + continue + if not result.ok: + await sender.resend(result) +``` + +For the legacy Bearer/HMAC path, pass the same registration to both helpers: +`deliver(registration, create_media_buy_delivery_webhook_payload(registration, task_id=..., result=...))`. +Durable outboxes can call `WebhookSender.prepare_media_buy_delivery(...)`, +persist the returned `PreparedWebhook`, and send it later with `send_prepared`. + +What the helpers guarantee: + +- `operation_id` and `token` are copied verbatim from the registration. The SDK + never derives an ID from the URL, `media_buy_id` or anything else. +- A registration without `operation_id`, or with `operation_id: null`, raises + `ReportingWebhookOperationIdMissingError` before any network I/O. The SDK also + logs one WARNING on the `adcp.webhooks` logger, with + `extra={"adcp_event": "reporting_webhook.operation_id_missing", "task_type": + "media_buy_delivery", "notification_type": ..., "media_buy_ids": (...)}`. The + URL and credentials are never logged. Use this record for a metric or alert, + and keep the exception for the per-registration flag. +- A stored `operation_id` that is not a 1–255 character `[A-Za-z0-9_.:-]` + string (for example, edited by hand, or never checked because SDK 8 accepts + it as an extra) raises `pydantic.ValidationError` instead of being sent. +- The envelope is `task_type="media_buy_delivery"` and `status="completed"`, + with your `task_id`. `notification_id` is omitted, because delivery reports + are point-in-time events deduped by `idempotency_key`. + `create_mcp_webhook_payload` no longer gives `media_buy_delivery` the + `"{task_id}.terminal"` default. + +### Migrating existing registrations + +Upgrading the SDK does not backfill seller storage or restart reports. +Registrations stored before buyers could send `operation_id` have none, and the +helpers skip them. Do not fill one in. Tell the affected buyers to update +`reporting_webhook` with an `operation_id`, through `update_media_buy` or +`control_media_buy`, and resume scheduled sends once the updated registration is +stored. Before deploying, test the whole path: buyer serialization, seller +validation, durable storage, scheduled send, and authenticated receipt. diff --git a/skills/build-seller-agent/SKILL.md b/skills/build-seller-agent/SKILL.md index 6d989ea3d..1e2186381 100644 --- a/skills/build-seller-agent/SKILL.md +++ b/skills/build-seller-agent/SKILL.md @@ -599,7 +599,8 @@ response.raise_for_status() ``` Notes: -- Copy `config.operation_id` verbatim into every MCP webhook payload. Do not derive it from the task, media buy, or callback URL. Legacy persisted reporting registrations without one must be flagged or skipped until the buyer supplies it. +- Copy `config.operation_id` verbatim into every MCP webhook payload. Do not derive it from the task, media buy, or callback URL. +- For scheduled delivery reports to a stored `reporting_webhook`, use `sender.send_media_buy_delivery(registration=stored, task_id=..., result=report)` (or `create_media_buy_delivery_webhook_payload` with `deliver`). It echoes the registration's `operation_id` and `token`, and raises `ReportingWebhookOperationIdMissingError` for a legacy registration without one. Catch it, skip that send, and ask the buyer to update the registration. See `docs/webhooks/reporting-operation-id-3.2.md`. - `deliver` hashes/signs the exact bytes it POSTs for HMAC-SHA256; for Bearer it attaches the credential as `Authorization`. Either way, the signer and the wire cannot disagree. - `deliver` emits a `DeprecationWarning` on first use; migrate to `WebhookSender` for 4.0. - If your buyer relies on `config.token` echo, pass `token_field="push_token"` (pick a name you and the receiver agree on — there is no spec-defined field name). diff --git a/src/adcp/webhook_sender.py b/src/adcp/webhook_sender.py index 20583f8a0..df4f04476 100644 --- a/src/adcp/webhook_sender.py +++ b/src/adcp/webhook_sender.py @@ -54,7 +54,10 @@ from adcp.types import ( AdcpProtocol, GeneratedTaskStatus, + McpWebhookPayload, + MediaBuyDeliveryWebhookResult, NotificationConfig, + ReportingWebhook, TaskType, WholesaleFeedEvent, WholesaleFeedWebhook, @@ -753,6 +756,14 @@ def prepare_mcp( idempotency_key=idempotency_key, token=token, ) + return WebhookSender._prepare_mcp_payload(url, payload, extra_headers) + + @staticmethod + def _prepare_mcp_payload( + url: str, + payload: McpWebhookPayload, + extra_headers: Mapping[str, str] | None, + ) -> PreparedWebhook: body_dict = { **to_wire_dict(payload), "idempotency_key": payload.idempotency_key, @@ -771,6 +782,73 @@ def prepare_mcp( extra_headers=dict(extra_headers) if extra_headers else {}, ) + async def send_media_buy_delivery( + self, + *, + registration: ReportingWebhook | Mapping[str, Any], + task_id: str, + result: MediaBuyDeliveryWebhookResult | dict[str, Any], + status: GeneratedTaskStatus | str = "completed", + timestamp: datetime | None = None, + message: str | None = None, + idempotency_key: str | None = None, + extra_headers: Mapping[str, str] | None = None, + ) -> WebhookDeliveryResult: + """Prepare and send one scheduled delivery report to a stored registration. + + Raises :class:`~adcp.webhooks.ReportingWebhookOperationIdMissingError` + without network I/O when the registration has no ``operation_id``. + """ + return await self.send_prepared( + self.prepare_media_buy_delivery( + registration=registration, + task_id=task_id, + result=result, + status=status, + timestamp=timestamp, + message=message, + idempotency_key=idempotency_key, + extra_headers=extra_headers, + ) + ) + + @staticmethod + def prepare_media_buy_delivery( + *, + registration: ReportingWebhook | Mapping[str, Any], + task_id: str, + result: MediaBuyDeliveryWebhookResult | dict[str, Any], + status: GeneratedTaskStatus | str = "completed", + timestamp: datetime | None = None, + message: str | None = None, + idempotency_key: str | None = None, + extra_headers: Mapping[str, str] | None = None, + ) -> PreparedWebhook: + """Prepare a ``media_buy_delivery`` webhook for a ``reporting_webhook``. + + The URL, ``operation_id`` and ``token`` all come from the same stored + registration, so the report echoes exactly what the buyer registered. + See :func:`~adcp.webhooks.create_media_buy_delivery_webhook_payload`. + + :raises ReportingWebhookOperationIdMissingError: the registration has + no ``operation_id``. Skip the send and flag it for the buyer. + """ + webhook = ( + registration + if isinstance(registration, ReportingWebhook) + else ReportingWebhook.model_validate(registration) + ) + payload = create_media_buy_delivery_webhook_payload( + webhook, + task_id=task_id, + result=result, + status=status, + timestamp=timestamp, + message=message, + idempotency_key=idempotency_key, + ) + return WebhookSender._prepare_mcp_payload(str(webhook.url), payload, extra_headers) + async def send_prepared( self, prepared: PreparedWebhook, @@ -1258,6 +1336,7 @@ async def _send_bytes( # cycle — see the note next to the removed top-level import. from adcp.webhooks import ( # noqa: E402 create_mcp_webhook_payload, + create_media_buy_delivery_webhook_payload, create_webhook_challenge_payload, generate_webhook_idempotency_key, to_wire_dict, diff --git a/src/adcp/webhooks.py b/src/adcp/webhooks.py index f6b6c9158..90ed61938 100644 --- a/src/adcp/webhooks.py +++ b/src/adcp/webhooks.py @@ -24,6 +24,7 @@ import hashlib import hmac import json +import logging import secrets import time import uuid @@ -31,7 +32,7 @@ from collections.abc import Mapping from dataclasses import dataclass from datetime import datetime, timezone -from typing import Any, NoReturn, cast +from typing import Annotated, Any, NoReturn, cast from urllib.parse import urlsplit import httpx @@ -42,7 +43,7 @@ ) from google.protobuf.json_format import MessageToDict, ParseDict from google.protobuf.struct_pb2 import Value -from pydantic import AnyUrl +from pydantic import AnyUrl, Field from pydantic import BaseModel as PydanticBaseModel from adcp.server.idempotency.backends import MemoryBackend as MemoryBackend @@ -61,7 +62,13 @@ verify_webhook_from_agent_url, verify_webhook_signature, ) -from adcp.types import AdcpProtocol, GeneratedTaskStatus, McpWebhookPayload, TaskType +from adcp.types import ( + AdcpProtocol, + GeneratedTaskStatus, + McpWebhookPayload, + ReportingWebhook, + TaskType, +) from adcp.types.base import AdCPBaseModel from adcp.webhook_receiver import ( LegacyHmacFallback, @@ -74,6 +81,8 @@ ) from adcp.webhook_transport_hooks import TransportHook, apply_hooks +logger = logging.getLogger(__name__) + # `task_type` → `protocol` mapping. Mirrors the JS reference # implementation's `TOOL_PROTOCOL_MAP` in # `adcontextprotocol/adcp-client:src/lib/server/decisioning/runtime/protocol-for-tool.ts` @@ -162,6 +171,8 @@ def create_mcp_webhook_payload( notification_id: Stable identity for one logical notification. Terminal task webhooks default to ``"{task_id}.terminal"`` so re-emissions under different delivery keys still converge. + ``media_buy_delivery`` reports get no default: they are + point-in-time events deduped by ``idempotency_key``. message: Human-readable summary of task state. context_id: Session/conversation identifier. protocol: AdCP protocol this task belongs to (see :class:`AdcpProtocol`). @@ -226,12 +237,14 @@ def create_mcp_webhook_payload( ) status_value = status.value if hasattr(status, "value") else str(status) - if notification_id is None and status_value in { - "completed", - "failed", - "canceled", - "rejected", - }: + task_type_value = task_type.value if hasattr(task_type, "value") else str(task_type) + # Delivery reports are point-in-time events, not terminal task outcomes: + # the spec omits notification_id and dedupes them by idempotency_key. + if ( + notification_id is None + and task_type_value != "media_buy_delivery" + and status_value in {"completed", "failed", "canceled", "rejected"} + ): notification_id = f"{task_id}.terminal" # Auto-derive `protocol` from `task_type` when caller doesn't override. @@ -279,6 +292,158 @@ def create_mcp_webhook_payload( return payload +#: Structured ``adcp_event`` value logged when a scheduled delivery report is +#: suppressed because its ``reporting_webhook`` registration has no +#: buyer-supplied ``operation_id``. +REPORTING_WEBHOOK_OPERATION_ID_MISSING = "reporting_webhook.operation_id_missing" + + +class ReportingWebhookOperationIdMissingError(ValueError): + """A ``reporting_webhook`` registration carries no buyer-supplied ``operation_id``. + + ``media_buy_delivery`` webhooks require ``operation_id``, and the seller + MUST echo the registration's value rather than invent one or parse it from + the URL. A registration without it cannot receive scheduled reports: skip + the send and ask the buyer to update ``reporting_webhook`` with an + ``operation_id``. ``get_media_buy_delivery`` stays available meanwhile. + + The SDK logs one WARNING on the ``adcp.webhooks`` logger before raising, + with ``extra`` fields ``adcp_event`` (:data:`REPORTING_WEBHOOK_OPERATION_ID_MISSING`), + ``task_type``, ``notification_type`` and ``media_buy_ids``. The URL and + credentials are never logged. + """ + + code = REPORTING_WEBHOOK_OPERATION_ID_MISSING + + def __init__(self, *, media_buy_ids: tuple[str, ...] = ()) -> None: + self.media_buy_ids = media_buy_ids + super().__init__( + "reporting_webhook has no buyer-supplied operation_id; media_buy_delivery " + "webhooks to this registration are suppressed. Ask the buyer to update " + "reporting_webhook.operation_id; never derive one from the URL or a " + "seller resource ID." + ) + + +def _delivery_report_media_buy_ids(result: Mapping[str, Any]) -> tuple[str, ...]: + deliveries = result.get("media_buy_deliveries") + if not isinstance(deliveries, list): + return () + return tuple( + delivery["media_buy_id"] + for delivery in deliveries + if isinstance(delivery, Mapping) and isinstance(delivery.get("media_buy_id"), str) + ) + + +# Same constraints as ``push_notification_config.operation_id``, which AdCP 3.3 +# adopts for ``reporting_webhook.operation_id`` (adcp#7895). SDK 8 is pinned to +# AdCP 3.2, whose ``ReportingWebhook`` does not declare the field; a +# buyer-supplied value survives as a model extra and is validated here, at +# send time, so registrations 8.x accepts today keep validating. +_REPORTING_OPERATION_ID_PATTERN = "^[A-Za-z0-9_.:-]{1,255}$" + + +class _ReportingWebhookOperationId(PydanticBaseModel): + operation_id: Annotated[ + str, + Field( + strict=True, + min_length=1, + max_length=255, + pattern=_REPORTING_OPERATION_ID_PATTERN, + ), + ] + + +def _reporting_webhook_operation_id(webhook: ReportingWebhook) -> str | None: + """Return the registration's buyer-supplied ``operation_id``, or ``None``. + + Reads the undeclared ``operation_id`` extra. Absent and ``None`` both mean + missing; any other value must be a 1–255 character ``[A-Za-z0-9_.:-]`` + string, or :class:`pydantic.ValidationError` is raised. + """ + value = (webhook.model_extra or {}).get("operation_id") + if value is None: + return None + return _ReportingWebhookOperationId.model_validate({"operation_id": value}).operation_id + + +def create_media_buy_delivery_webhook_payload( + registration: ReportingWebhook | Mapping[str, Any], + *, + task_id: str, + result: PydanticBaseModel | dict[str, Any], + status: GeneratedTaskStatus | str = "completed", + timestamp: datetime | None = None, + message: str | None = None, + idempotency_key: str | None = None, +) -> McpWebhookPayload: + """Build a scheduled ``media_buy_delivery`` webhook for a stored registration. + + ``operation_id`` and ``token`` are copied verbatim from ``registration`` + (a :class:`ReportingWebhook` or its persisted JSON form), so the reporting + stream correlates with the value the buyer registered. ``notification_id`` + is omitted: delivery reports are deduped by ``idempotency_key``. + + AdCP 3.2's ``ReportingWebhook`` does not declare ``operation_id``, so SDK 8 + reads the buyer's value from the model's extra fields (AdCP 3.3 adds the + field with these constraints; adcp#7895). + + Args: + registration: The buyer's ``reporting_webhook`` as stored by the seller. + task_id: Seller-generated identifier for this report, for example + ``"delivery_report_67_2026_04_000031"``. + result: The ``MediaBuyDeliveryWebhookResult`` body, kept byte-for-byte. + status: Envelope status, normally ``"completed"``. It is distinct from + ``media_buy_deliveries[].status``. + timestamp: When the report was first generated. Defaults to now (UTC). + message: Optional human-readable summary. + idempotency_key: The first attempt's key when rebuilding for a retry. + + Raises: + ReportingWebhookOperationIdMissingError: ``registration`` has no + ``operation_id``, or it is ``None``. Skip this send; do not + fabricate a value. + pydantic.ValidationError: ``registration`` is malformed, including a + stored ``operation_id`` that is not a 1–255 character + ``[A-Za-z0-9_.:-]`` string. + """ + webhook = ( + registration + if isinstance(registration, ReportingWebhook) + else ReportingWebhook.model_validate(registration) + ) + operation_id = _reporting_webhook_operation_id(webhook) + if operation_id is None: + result_dict = ( + result.model_dump(mode="json") if isinstance(result, PydanticBaseModel) else result + ) + media_buy_ids = _delivery_report_media_buy_ids(result_dict) + logger.warning( + "Suppressed media_buy_delivery webhook: reporting_webhook has no " + "buyer-supplied operation_id", + extra={ + "adcp_event": REPORTING_WEBHOOK_OPERATION_ID_MISSING, + "task_type": "media_buy_delivery", + "notification_type": result_dict.get("notification_type"), + "media_buy_ids": media_buy_ids, + }, + ) + raise ReportingWebhookOperationIdMissingError(media_buy_ids=media_buy_ids) + return create_mcp_webhook_payload( + task_id=task_id, + status=status, + task_type=TaskType.media_buy_delivery, + result=result, + timestamp=timestamp, + operation_id=operation_id, + message=message, + idempotency_key=idempotency_key, + token=webhook.token, + ) + + def get_adcp_signed_headers_for_webhook( headers: dict[str, Any], secret: str, @@ -2025,6 +2190,9 @@ def _validate_header_value(name: str, value: Any) -> None: # Sender — payload builders "create_a2a_webhook_payload", "create_mcp_webhook_payload", + "create_media_buy_delivery_webhook_payload", + "REPORTING_WEBHOOK_OPERATION_ID_MISSING", + "ReportingWebhookOperationIdMissingError", "generate_webhook_idempotency_key", "generate_webhook_challenge_value", "create_webhook_challenge_payload", diff --git a/tests/test_reporting_webhook_operation_id.py b/tests/test_reporting_webhook_operation_id.py index 8f3c70fdc..21c498114 100644 --- a/tests/test_reporting_webhook_operation_id.py +++ b/tests/test_reporting_webhook_operation_id.py @@ -1,16 +1,46 @@ -"""Reporting webhooks stay separate from MCP envelope correlation.""" +"""Reporting-webhook operation_id: registration extra and scheduled delivery. + +AdCP 3.2 does not declare ``reporting_webhook.operation_id``; SDK 8 keeps a +buyer-supplied value as a model extra and validates it when a delivery report +is built. AdCP 3.3 standardizes the field (adcp#7895). +""" from __future__ import annotations import json +import logging from pathlib import Path +from typing import Any +import httpx import pytest from pydantic import ValidationError -from adcp.types import McpWebhookPayload, ReportingWebhook +from adcp.types import ( + AcceptProposalRequest as CurrentAcceptProposalRequest, +) +from adcp.types import ( + BuyProductsRequest, + ControlMediaBuyRequest, + CreateMediaBuyRequest, + McpWebhookPayload, + MediaBuyDeliveryWebhookResult, + PushNotificationConfig, + ReportingWebhook, + UpdateMediaBuyRequest, +) from adcp.types.v32 import AcceptProposalRequest +from adcp.validation.schema_loader import get_named_validator from adcp.validation.version import resolve_bundle_key +from adcp.webhooks import ( + REPORTING_WEBHOOK_OPERATION_ID_MISSING, + ReportingWebhookOperationIdMissingError, + WebhookSender, + _ReportingWebhookOperationId, + create_mcp_webhook_payload, + create_media_buy_delivery_webhook_payload, + deliver, +) _AUTHENTICATION = { "schemes": ["Bearer"], @@ -31,6 +61,333 @@ def test_reporting_webhook_does_not_define_operation_id() -> None: assert "operation_id" not in webhook.model_dump() +def test_buyer_supplied_operation_id_survives_as_extra() -> None: + """No typed field on 8.x: the value round-trips as an extra, unvalidated.""" + webhook = ReportingWebhook( + url="https://buyer.example/reporting", + authentication=_AUTHENTICATION, + reporting_frequency="daily", + operation_id="op_reporting_001", + ) + assert "operation_id" not in ReportingWebhook.model_fields + assert webhook.model_extra == {"operation_id": "op_reporting_001"} + restored = ReportingWebhook.model_validate_json(webhook.model_dump_json()) + assert restored.model_extra == {"operation_id": "op_reporting_001"} + + +_ACCOUNT = {"brand": {"domain": "example.com"}, "operator": "agency.example"} +_REQUEST_CASES = [ + ( + CurrentAcceptProposalRequest, + {"proposal_id": "proposal-1", "proposal_terms_digest": "sha256:" + "x" * 43}, + ), + ( + BuyProductsRequest, + { + "feed_version": "feed-1", + "purchases": [{"product_id": "product-1", "pricing_option_id": "price-1"}], + "start_time": "2026-10-01T00:00:00Z", + "end_time": "2026-10-31T23:59:59Z", + }, + ), + (ControlMediaBuyRequest, {"media_buy_id": "buy-1", "revision": 1}), + ( + CreateMediaBuyRequest, + { + "brand": {"domain": "example.com"}, + "start_time": "2026-10-01T00:00:00Z", + "end_time": "2026-10-31T23:59:59Z", + "packages": [ + {"product_id": "product-1", "pricing_option_id": "price-1", "budget": 100.0} + ], + }, + ), + (UpdateMediaBuyRequest, {"media_buy_id": "buy-1"}), +] + + +@pytest.mark.parametrize("model,params", _REQUEST_CASES) +def test_registration_requests_carry_operation_id_to_the_delivery_helper( + model: Any, params: dict[str, Any] +) -> None: + """The extra survives each registering request and its stored JSON form.""" + request = model( + **params, + adcp_version="3.2", + account=_ACCOUNT, + idempotency_key="reporting_request_001", + reporting_webhook={ + "url": "https://buyer.example/reporting", + "authentication": _AUTHENTICATION, + "reporting_frequency": "daily", + "operation_id": "op_reporting_001", + }, + ) + stored = model.model_validate_json(request.model_dump_json()).reporting_webhook + payload = create_media_buy_delivery_webhook_payload( + stored, task_id="delivery_report_000001", result=_DELIVERY_REPORT + ) + assert payload.operation_id == "op_reporting_001" + + +_DELIVERY_REPORT = { + "notification_type": "scheduled", + "sequence_number": 3, + "reporting_period": { + "start": "2026-10-01T00:00:00Z", + "end": "2026-10-02T00:00:00Z", + }, + "currency": "USD", + "media_buy_deliveries": [ + { + "media_buy_id": "mb_001", + "status": "active", + "totals": {"impressions": 1000, "spend": 45.0}, + "by_package": [], + } + ], +} + + +def _validate_wire_delivery_webhook(body: bytes) -> dict[str, Any]: + """Validate the posted bytes against the pinned bundle, envelope and result.""" + wire = json.loads(body) + for path, value in ( + ("core/mcp-webhook-payload.json", wire), + ("media-buy/media-buy-delivery-webhook-result.json", wire["result"]), + ): + validator = get_named_validator(path, version="3.2") + assert validator is not None + validator.validate(value) + return wire + + +def test_reporting_operation_id_constraints_match_push_notification_config() -> None: + """AdCP 3.3 (adcp#7895) adopts push_notification_config.operation_id's constraints. + + SDK 8 has no typed field, so the delivery helpers validate the extra + themselves. Pin parity with the generated push-notification field. + """ + + def constraints(field: Any) -> dict[str, Any]: + found: dict[str, Any] = {} + for item in field.metadata: + for name in ("min_length", "max_length", "pattern"): + if getattr(item, name, None) is not None: + found[name] = getattr(item, name) + return found + + push = PushNotificationConfig.model_fields["operation_id"] + reporting = _ReportingWebhookOperationId.model_fields["operation_id"] + assert ( + constraints(reporting) + == constraints(push) + == { + "min_length": 1, + "max_length": 255, + "pattern": "^[A-Za-z0-9_.:-]{1,255}$", + } + ) + + +@pytest.mark.asyncio +async def test_stored_registration_correlates_delivery_and_retry(tmp_path: Path) -> None: + registration = ReportingWebhook( + url="https://buyer.example/reporting?operation_id=do-not-parse-this", + authentication=_AUTHENTICATION, + reporting_frequency="daily", + operation_id="op_reporting_001", + token="buyer-reporting-echo-token", + ) + stored = tmp_path / "registration.json" + stored.write_text(registration.model_dump_json()) + restored = json.loads(stored.read_text()) + received: list[bytes] = [] + + def receive(request: httpx.Request) -> httpx.Response: + assert str(request.url) == str(registration.url) + assert request.headers["authorization"] == "Bearer buyer-reporting-token-1234567890" + payload = McpWebhookPayload.model_validate_json(request.content) + assert payload.operation_id == "op_reporting_001" + assert payload.task_type.value == "media_buy_delivery" + assert payload.status.value == "completed" + received.append(request.content) + return httpx.Response(200) + + async with httpx.AsyncClient(transport=httpx.MockTransport(receive)) as client: + async with WebhookSender.from_bearer_token( + _AUTHENTICATION["credentials"], client=client + ) as sender: + prepared = sender.prepare_media_buy_delivery( + registration=restored, + task_id="delivery_report_mb_001_000003", + result=_DELIVERY_REPORT, + ) + assert (await sender.send_prepared(prepared)).ok + assert (await sender.send_prepared(prepared)).ok + assert received[0] == received[1] + wire = _validate_wire_delivery_webhook(received[0]) + assert wire["operation_id"] == "op_reporting_001" + assert wire["token"] == "buyer-reporting-echo-token" + assert wire["result"] == _DELIVERY_REPORT + assert wire["idempotency_key"] != wire["operation_id"] + # Point-in-time delivery events carry no notification_id. + assert "notification_id" not in wire + + +@pytest.mark.asyncio +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +async def test_legacy_deliver_echoes_registered_operation_id() -> None: + registration = ReportingWebhook( + url="https://buyer.example/reporting", + authentication=_AUTHENTICATION, + reporting_frequency="daily", + operation_id="op_reporting_legacy", + ) + received: list[httpx.Request] = [] + + def receive(request: httpx.Request) -> httpx.Response: + received.append(request) + return httpx.Response(204) + + async with httpx.AsyncClient(transport=httpx.MockTransport(receive)) as client: + response = await deliver( + registration, + create_media_buy_delivery_webhook_payload( + registration, task_id="delivery_report_000001", result=_DELIVERY_REPORT + ), + client=client, + ) + assert response.status_code == 204 + wire = _validate_wire_delivery_webhook(received[0].content) + assert wire["operation_id"] == "op_reporting_legacy" + assert received[0].headers["authorization"] == f"Bearer {_AUTHENTICATION['credentials']}" + + +@pytest.mark.asyncio +async def test_scheduled_sends_skip_and_flag_registrations_without_operation_id( + caplog: pytest.LogCaptureFixture, +) -> None: + """A legacy registration is suppressed and flagged, never given a minted ID.""" + legacy = { + "url": "https://buyer.example/reporting/mb_legacy?op=not-a-source", + "authentication": _AUTHENTICATION, + "reporting_frequency": "daily", + } + current = {**legacy, "url": "https://buyer.example/reporting", "operation_id": "op_current"} + received: list[dict[str, Any]] = [] + skipped: list[ReportingWebhookOperationIdMissingError] = [] + + def receive(request: httpx.Request) -> httpx.Response: + received.append(json.loads(request.content)) + return httpx.Response(200) + + caplog.set_level(logging.WARNING, logger="adcp.webhooks") + async with httpx.AsyncClient(transport=httpx.MockTransport(receive)) as client: + async with WebhookSender.from_bearer_token( + _AUTHENTICATION["credentials"], client=client + ) as sender: + # A seller's scheduler loop over stored registrations. + for media_buy_id, stored in (("mb_legacy", legacy), ("mb_current", current)): + report = { + **_DELIVERY_REPORT, + "media_buy_deliveries": [ + { + **_DELIVERY_REPORT["media_buy_deliveries"][0], + "media_buy_id": media_buy_id, + } + ], + } + try: + result = await sender.send_media_buy_delivery( + registration=stored, + task_id=f"delivery_report_{media_buy_id}", + result=report, + ) + except ReportingWebhookOperationIdMissingError as exc: + skipped.append(exc) + continue + assert result.ok + + assert [payload["operation_id"] for payload in received] == ["op_current"] + assert len(skipped) == 1 + assert skipped[0].code == REPORTING_WEBHOOK_OPERATION_ID_MISSING + assert skipped[0].media_buy_ids == ("mb_legacy",) + [record] = [r for r in caplog.records if r.name == "adcp.webhooks"] + assert record.levelno == logging.WARNING + assert record.adcp_event == REPORTING_WEBHOOK_OPERATION_ID_MISSING # type: ignore[attr-defined] + assert record.task_type == "media_buy_delivery" # type: ignore[attr-defined] + assert record.notification_type == "scheduled" # type: ignore[attr-defined] + assert record.media_buy_ids == ("mb_legacy",) # type: ignore[attr-defined] + rendered = record.getMessage() + str(skipped[0]) + assert "buyer.example" not in rendered + assert _AUTHENTICATION["credentials"] not in rendered + + +@pytest.mark.parametrize("extra", [{}, {"operation_id": None}]) +def test_typed_registration_without_operation_id_is_refused(extra: dict[str, Any]) -> None: + registration = ReportingWebhook( + url="https://buyer.example/reporting", + authentication=_AUTHENTICATION, + reporting_frequency="daily", + **extra, + ) + with pytest.raises(ReportingWebhookOperationIdMissingError): + create_media_buy_delivery_webhook_payload( + registration, + task_id="delivery_report_000001", + result=MediaBuyDeliveryWebhookResult.model_validate(_DELIVERY_REPORT), + ) + + +@pytest.mark.parametrize("operation_id", ["", "a" * 256, "has spaces", "a/b", "a\n", 123]) +def test_tampered_stored_operation_id_fails_validation(operation_id: object) -> None: + with pytest.raises(ValidationError, match="operation_id"): + WebhookSender.prepare_media_buy_delivery( + registration={ + "url": "https://buyer.example/reporting", + "authentication": _AUTHENTICATION, + "reporting_frequency": "daily", + "operation_id": operation_id, + }, + task_id="delivery_report_000001", + result=_DELIVERY_REPORT, + ) + + +@pytest.mark.parametrize("operation_id", ["op_reporting_001", "a" * 255, "buyer.stream:1-2"]) +def test_valid_operation_id_extra_is_echoed(operation_id: str) -> None: + payload = create_media_buy_delivery_webhook_payload( + { + "url": "https://buyer.example/reporting", + "authentication": _AUTHENTICATION, + "reporting_frequency": "daily", + "operation_id": operation_id, + }, + task_id="delivery_report_000001", + result=_DELIVERY_REPORT, + ) + assert payload.operation_id == operation_id + + +def test_generic_mcp_builder_omits_terminal_id_only_for_delivery_reports() -> None: + delivery = create_mcp_webhook_payload( + task_id="delivery_report_000001", + status="completed", + task_type="media_buy_delivery", + operation_id="op_reporting_001", + result=_DELIVERY_REPORT, + ) + task = create_mcp_webhook_payload( + task_id="task_001", + status="completed", + task_type="create_media_buy", + operation_id="op_task_001", + ) + assert delivery.notification_id is None + assert task.notification_id == "task_001.terminal" + + def test_beta6_versioned_request_schema_preserves_reporting_operation_id() -> None: """Historical versioned types retain the beta.6 wire contract.""" with pytest.raises(ValidationError, match="operation_id.*required property"):