From 99868a1b77b7f53c34f1e132976373b95ede879a Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 22:23:27 +0000 Subject: [PATCH] feat(server): seller helper for wholesale feed versions Backport of #1706 to SDK 9.x (AdCP 3.2), whose get_products / get_signals wholesale fields, if_wholesale_feed_version / if_pricing_version rules and unchanged: true response are the same as 3.3's. adcp.server gains wholesale_feed_version and pricing_version (RFC 8785 canonical, SHA-256, order-independent across items, path-addressed freshness fields ignored, keyed on cache_scope), check_feed_preconditions (the mode and pairing rules, INVALID_REQUEST with field) and feed_unchanged. products_response / signals_response no longer write unchanged: false, and refuse unchanged=True with items or without wholesale_feed_version, both of which the schema already rejects. Differences from #1706: the extra= docs no longer name the 3.3-only collections / signal_pricing_group_rules, and one test compares a model with its own wire form, since SDK 9 dumps unset schema defaults. Refs #1676 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/wholesale-feed-versions.md | 145 ++++++++ src/adcp/server/__init__.py | 15 + src/adcp/server/feed_version.py | 362 +++++++++++++++++++ src/adcp/server/responses.py | 37 +- tests/test_feed_version.py | 600 ++++++++++++++++++++++++++++++++ 5 files changed, 1154 insertions(+), 5 deletions(-) create mode 100644 docs/wholesale-feed-versions.md create mode 100644 src/adcp/server/feed_version.py create mode 100644 tests/test_feed_version.py diff --git a/docs/wholesale-feed-versions.md b/docs/wholesale-feed-versions.md new file mode 100644 index 000000000..09a3d004d --- /dev/null +++ b/docs/wholesale-feed-versions.md @@ -0,0 +1,145 @@ +# Wholesale feed versions (seller side) + +A wholesale `get_products` (`buying_mode="wholesale"`) or `get_signals` +(`discovery_mode="wholesale"`) response carries an opaque +`wholesale_feed_version`. A buyer such as `adcp.FeedMirror` echoes it back as +`if_wholesale_feed_version` and expects `unchanged: true`, with the item list +omitted, when nothing it can see has changed. `adcp.server.feed_version` +provides the seller side: + +| Helper | Job | +| --- | --- | +| `wholesale_feed_version(items, *, cache_scope, ignore=FRESHNESS_FIELDS, extra=None)` | Content token for the feed | +| `pricing_version(items, *, cache_scope)` | Token for the pricing layer, for sellers that version pricing separately | +| `check_feed_preconditions(request)` | Mode and pairing rules for `if_*` tokens. Raises `AdcpError("INVALID_REQUEST", field=...)` | +| `feed_unchanged(request, *, feed_version, pricing_version=None)` | Returns whether the buyer's tokens are current | + +All four are exported from `adcp.server`. + +```python +from adcp.server import feed_unchanged, products_response, wholesale_feed_version + +async def get_products(params, context=None): + scope = "account" if account_has_overrides(params.account) else "public" + feed = await load_wholesale_feed(params, scope) # the whole feed, not one page + version = wholesale_feed_version(feed, cache_scope=scope) + if feed_unchanged(params, feed_version=version): # also runs check_feed_preconditions + return products_response( + unchanged=True, wholesale_feed_version=version, cache_scope=scope + ) + page, pagination = paginate(feed, params.pagination) + return products_response( + page, wholesale_feed_version=version, cache_scope=scope, pagination=pagination + ) +``` + +`get_signals` follows the same pattern with `signals_response`. + +## The token + +The token is `"sha256:"` followed by 64 hex digits. The schema places no +format or length constraint on it, and buyers must treat it as opaque. Equal +buyer-visible content always produces the same token. Any buyer-visible change +produces a different one. + +- **Wire form.** Each item, whether a dict or a Pydantic model, is serialized + the way `products_response` / `signals_response` puts it on the wire. It is + then RFC 8785 (JCS) canonicalized and hashed with SHA-256. +- **Order.** Only the order of top-level items is ignored, because a feed is a + set keyed by `product_id` / `signal_agent_segment_id`. Arrays inside an item + keep their order: some are ordered by contract (`forecast.points` ascend by + budget), and the schema declares no set semantics for the rest. A seller + that emits a stable order gets stable tokens. A reordering costs at most one spurious + refetch, while a missed change would leave the buyer with stale data. +- **Freshness fields.** `ignore` takes dot paths from the item root, and + arrays are traversed without a path segment. `FRESHNESS_FIELDS` is + `forecast.generated_at` (products) and `coverage_forecast.generated_at` + (signals): a regenerated forecast with the same numbers is not a change. + `valid_until` is still hashed, because buyers read it to decide whether a + forecast has expired. A bare key matches only at the item root, never by + name anywhere in the item. +- **Scope.** `cache_scope` is part of the hash. The spec keys the comparator + on `(cache_scope, wholesale_feed_version)`, so a public token never matches + an account-scoped read, even when the content is identical. +- **Pages.** `pagination.cursor` is outside the version's scope. Hash the + whole feed and put the same token on every page. `feed_unchanged` returns + `False` on continuation pages. Hashing is linear in the feed size, so a + seller with a large feed computes the token when the feed changes and + stores it alongside the feed. +- **Response-level content.** Pass `extra` for content outside the item list + that the buyer sees and that must move the version. AdCP 3.2 responses + carry none; it exists for AdCP 3.3's `collections` (SDK 10). An empty value + still produces a different token from no `extra`. + +## Separate pricing version + +`pricing_version` covers `PRICING_FIELDS`, which are top-level +`pricing_options` and `signal_targeting_options[].pricing_options`, the +fields `GetProductsResponse.pricing_version` names. It binds each price to its +item's ID. To separate pricing: + +```python +feed_token = wholesale_feed_version(feed, cache_scope=scope, + ignore=FRESHNESS_FIELDS | PRICING_FIELDS) +price_token = pricing_version(feed, cache_scope=scope) +if feed_unchanged(params, feed_version=feed_token, pricing_version=price_token): + return products_response(unchanged=True, wholesale_feed_version=feed_token, + pricing_version=price_token, cache_scope=scope) +``` + +`feed_unchanged` follows the spec's evaluation order: a feed mismatch returns +the full payload, and so does a matching feed with a pricing mismatch. Only +when both match is the answer `unchanged`. If the seller passes +`pricing_version` and the buyer did not send `if_pricing_version`, the answer +is the full payload, because the feed token alone cannot show that prices are +current. Separating pricing is optional. A wholesale feed webhook carries +only `wholesale_feed_version`, and with separate pricing that token does not +move on a price-only change. + +## Request rules + +`check_feed_preconditions` rejects the following with `INVALID_REQUEST`: + +- `if_pricing_version` without `if_wholesale_feed_version`, with + `field="if_pricing_version"`. +- Either token on a read that is not wholesale, with + `field="if_wholesale_feed_version"`. A `get_signals` request without + `discovery_mode` is a brief read. + +It accepts request models and wire dicts. It complements +`adcp.decisioning.refine.assert_buying_mode_consistent`, which checks `brief` +and `refine` but not these tokens. + +## The unchanged answer + +`products_response(unchanged=True, ...)` and `signals_response(unchanged=True, ...)` +omit the item list and require `wholesale_feed_version`. Pass `cache_scope` +too: the schema requires it, and the MCP dispatcher fills in `public` only +when the request has no account. Echo `pricing_version` when you use it. +Passing items with `unchanged=True`, or leaving out the version, raises +`ValueError`. `unchanged=False` +emits nothing, because the spec forbids `unchanged: false`: a response that +carries products simply has no `unchanged` field. + +## Webhooks + +After a change, pass the post-change token to the wholesale feed webhook: + +```python +await sender.send_wholesale_feed( + ..., + wholesale_feed_version=wholesale_feed_version(feed_after, cache_scope="public"), + previous_wholesale_feed_version=token_before, + cache_scope="public", +) +``` + +A mirror that applies the webhook stores this version and presents it on its +next conditional read. Since the read computes the same token from the same +feed, the seller answers `unchanged: true`. A buyer that uses the version as a +gap detector therefore sees no mismatch. + +`list_products` returns `feed_version` and accepts `if_feed_version`, the +same concept under a different name. The token functions work for it +unchanged. `feed_unchanged` and the response builders cover `get_products` +and `get_signals`. diff --git a/src/adcp/server/__init__.py b/src/adcp/server/__init__.py index b9ef2b5d6..2428301ce 100644 --- a/src/adcp/server/__init__.py +++ b/src/adcp/server/__init__.py @@ -103,6 +103,14 @@ async def get_products(params, context=None): build_manifest, make_discovery_route, ) +from adcp.server.feed_version import ( + FRESHNESS_FIELDS, + PRICING_FIELDS, + check_feed_preconditions, + feed_unchanged, + pricing_version, + wholesale_feed_version, +) from adcp.server.governance import GovernanceHandler from adcp.server.governance_enforcement import ( GovernanceEnforcementDecision, @@ -364,4 +372,11 @@ async def get_products(params, context=None): "sync_creatives_response", "sync_governance_response", "update_media_buy_response", + # Wholesale feed versions (conditional get_products / get_signals reads) + "FRESHNESS_FIELDS", + "PRICING_FIELDS", + "check_feed_preconditions", + "feed_unchanged", + "pricing_version", + "wholesale_feed_version", ] diff --git a/src/adcp/server/feed_version.py b/src/adcp/server/feed_version.py new file mode 100644 index 000000000..892a4eace --- /dev/null +++ b/src/adcp/server/feed_version.py @@ -0,0 +1,362 @@ +"""Seller half of conditional wholesale feed reads. + +A wholesale ``get_products`` (``buying_mode='wholesale'``) or ``get_signals`` +(``discovery_mode='wholesale'``) response carries an opaque +``wholesale_feed_version`` (and optionally ``pricing_version``). A buyer echoes +it as ``if_wholesale_feed_version`` / ``if_pricing_version``; when nothing has +changed the seller answers ``unchanged: true`` with the item list omitted. +:mod:`adcp.feed_mirror` is the buyer half. This module gives the seller the +three pieces it otherwise hand-rolls: + +- :func:`wholesale_feed_version` / :func:`pricing_version` — content tokens. + Same buyer-visible content, same token; any buyer-visible change, a new one. +- :func:`check_feed_preconditions` — the request's mode and pairing rules. +- :func:`feed_unchanged` — whether the buyer's tokens are current. Answer + with ``products_response(unchanged=True, ...)`` / + ``signals_response(unchanged=True, ...)``. + +Example:: + + from adcp.server import feed_unchanged, products_response, wholesale_feed_version + + async def get_products(params, context=None): + scope = "account" if has_overrides(params.account) else "public" + feed = await load_wholesale_feed(params, scope) # the whole feed, every page + version = wholesale_feed_version(feed, cache_scope=scope) + if feed_unchanged(params, feed_version=version): + return products_response( + unchanged=True, wholesale_feed_version=version, cache_scope=scope + ) + page, pagination = paginate(feed, params.pagination) + return products_response( + page, wholesale_feed_version=version, cache_scope=scope, pagination=pagination + ) + +Pass the same token as ``wholesale_feed_version`` to +:meth:`adcp.webhooks.WebhookSender.send_wholesale_feed` after a change, so a +buyer that uses the webhook version as a gap detector sees the version its +next conditional read returns. +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Iterable, Mapping +from typing import TYPE_CHECKING, Any + +import rfc8785 +from pydantic_core import to_jsonable_python + +from adcp.server.responses import _serialize + +if TYPE_CHECKING: + from adcp.types import GetProductsRequest, GetSignalsRequest + +__all__ = [ + "FRESHNESS_FIELDS", + "PRICING_FIELDS", + "check_feed_preconditions", + "feed_unchanged", + "pricing_version", + "wholesale_feed_version", +] + +# Item-relative paths of fields that record when a value was computed rather +# than what it is. A regenerated forecast with the same numbers is not a change +# a buyer needs to refetch for. ``valid_until`` stays in: a buyer reads it to +# decide whether its mirrored forecast has expired. +FRESHNESS_FIELDS: frozenset[str] = frozenset( + { + "forecast.generated_at", # Product.forecast (DeliveryForecast) + "coverage_forecast.generated_at", # get_signals signals[].coverage_forecast + } +) + +# The pricing layer, per GetProductsResponse.pricing_version: "product +# pricing_options and nested signal_targeting_options pricing_options". +# Signals carry ``pricing_options`` at the top level of each item. +PRICING_FIELDS: frozenset[str] = frozenset( + { + "pricing_options", + "signal_targeting_options.pricing_options", + } +) + +# Stable item identity: the keys FeedMirror indexes products and signals by. +_ITEM_ID_FIELDS = ("product_id", "signal_agent_segment_id") + +_CACHE_SCOPES = ("public", "account") + +_Path = tuple[str, ...] + + +def wholesale_feed_version( + items: Iterable[Any], + *, + cache_scope: str, + ignore: Iterable[str] = FRESHNESS_FIELDS, + extra: Mapping[str, Any] | None = None, +) -> str: + """Return a content token for a wholesale product or signal feed. + + ``items`` is the *whole* feed for the request's scoping tuple, not one + page: ``pagination.cursor`` is outside the version's scope, so every page + carries the same token. Items may be dicts or Pydantic models; each is + serialized exactly as :func:`~adcp.server.responses.products_response` + puts it on the wire, so the token follows what the buyer sees. + + Canonicalization: + + - Each item is RFC 8785 (JCS) canonical JSON — sorted keys, no + whitespace, ES6 number form — hashed with SHA-256. The token is the + SHA-256 of the sorted item digests, so the order of top-level items + does not matter (a feed is a set keyed by ``product_id`` / + ``signal_agent_segment_id``). Arrays *inside* an item keep their + order: some are ordered by contract (``forecast.points`` ascend by + budget), the schema declares no set semantics for the rest, and a + reordering costs at worst one spurious refetch. + - ``ignore`` names fields left out of the hash, as dot-separated paths + from the item root (``"forecast.generated_at"``). Arrays are traversed + without a path segment, so ``"pricing_options.price_guidance"`` + reaches every pricing option. A bare key matches only at the item root, + never by name anywhere. Default: :data:`FRESHNESS_FIELDS`. A seller that + versions pricing separately passes ``FRESHNESS_FIELDS | PRICING_FIELDS`` + and also returns :func:`pricing_version`. + - ``cache_scope`` is part of the hash. The spec keys the comparator on + ``(cache_scope, wholesale_feed_version)``; a public token can then never + match an account-scoped read with the same content, or vice versa. + - ``extra`` carries response-level content the buyer sees beside the + items and that must move the version (an empty value still differs + from ``None``). AdCP 3.2 responses have none; SDK 10 uses it for 3.3's + ``collections``. It is hashed as-is, without ``ignore``. + + :returns: ``"sha256:"`` followed by 64 hex digits. The schema puts no + format, length, or pattern constraint on the token; buyers treat it as + opaque. + :raises ValueError: an unknown ``cache_scope``, a malformed ``ignore`` + path, or a value outside I-JSON (for example a NaN or an integer + beyond 2**53). + """ + paths = _parse_paths(ignore) + digests = sorted(_digest(_without(item, paths)) for item in _wire_items(items)) + return _token(cache_scope, digests, extra) + + +def pricing_version(items: Iterable[Any], *, cache_scope: str) -> str: + """Return a token for the pricing layer of a wholesale feed. + + Covers :data:`PRICING_FIELDS` of every item, bound to the item by its + ``product_id`` / ``signal_agent_segment_id`` so moving a price from one + product to another is a change. Use it with + ``wholesale_feed_version(items, ignore=FRESHNESS_FIELDS | PRICING_FIELDS)``: + a re-price then moves only this token, and a structural change moves the + feed token (which forces a full payload on its own). Canonicalization and + ``cache_scope`` keying are as for :func:`wholesale_feed_version`. + + Separate pricing is optional; the spec lets a seller omit + ``pricing_version`` and let ``wholesale_feed_version`` cover both. Note + that a wholesale feed webhook carries only ``wholesale_feed_version``, + which a price-only change does not move when pricing is separated. + + :raises ValueError: an item with neither identity field, or as for + :func:`wholesale_feed_version`. + """ + keep = _parse_paths(PRICING_FIELDS) + digests = [] + for item in _wire_items(items): + if not isinstance(item, dict) or not any(key in item for key in _ITEM_ID_FIELDS): + raise ValueError( + "pricing_version items need product_id or signal_agent_segment_id " + "to bind each price to its item" + ) + projection = {key: item[key] for key in _ITEM_ID_FIELDS if key in item} + projection.update(_only(item, keep)) + digests.append(_digest(projection)) + return _token(cache_scope, sorted(digests), None) + + +def check_feed_preconditions(request: GetProductsRequest | GetSignalsRequest | Any) -> None: + """Enforce the conditional-read rules of a get_products / get_signals request. + + - ``if_pricing_version`` requires ``if_wholesale_feed_version`` (pricing + has no structural baseline on its own). + - Either token requires a wholesale read: ``buying_mode='wholesale'`` on + get_products, ``discovery_mode='wholesale'`` on get_signals. + + ``request`` is a request model or its wire dict. A request without + either token passes. Complements + :func:`adcp.decisioning.refine.assert_buying_mode_consistent`, which + checks ``brief`` / ``refine`` but not these tokens. + + :raises adcp.decisioning.AdcpError: ``INVALID_REQUEST`` with ``field`` + naming the offending token. + """ + from adcp.decisioning.types import AdcpError + + feed_token = _get(request, "if_wholesale_feed_version") + pricing_token = _get(request, "if_pricing_version") + if pricing_token is not None and feed_token is None: + raise AdcpError( + "INVALID_REQUEST", + message=( + "if_pricing_version must be sent together with " + "if_wholesale_feed_version; a pricing version has no " + "structural baseline to compare against on its own." + ), + recovery="correctable", + field="if_pricing_version", + ) + if feed_token is None: + return + mode_field, mode = _mode(request) + if mode != "wholesale": + raise AdcpError( + "INVALID_REQUEST", + message=( + f"if_wholesale_feed_version is only valid on a wholesale feed read " + f"({mode_field}='wholesale'); got {mode_field}={mode!r}." + ), + recovery="correctable", + field="if_wholesale_feed_version", + ) + + +def feed_unchanged( + request: GetProductsRequest | GetSignalsRequest | Any, + *, + feed_version: str, + pricing_version: str | None = None, +) -> bool: + """Return whether the buyer's cached feed is current. + + ``feed_version`` / ``pricing_version`` are the seller's current tokens + for the cache_scope it resolved *this* request to (from + :func:`wholesale_feed_version` / :func:`pricing_version`, which key the + token on that scope). Pass ``pricing_version`` only when the seller + versions pricing separately. + + True only when ``if_wholesale_feed_version`` matches and — when the + seller passes ``pricing_version`` — ``if_pricing_version`` is present and + matches. A buyer that omits ``if_pricing_version`` from a seller that + separates pricing gets the full payload: its feed token alone cannot show + that prices are current. A seller that does not track pricing (no + ``pricing_version``) ignores ``if_pricing_version``, as the spec allows. + A continuation page (``pagination.cursor`` set) is never unchanged — the + tokens version the feed's first page read. + + On True, answer with ``products_response(unchanged=True, + wholesale_feed_version=..., cache_scope=..., pricing_version=...)`` (or + :func:`~adcp.server.responses.signals_response`), echoing + ``pricing_version`` when used. + + :raises adcp.decisioning.AdcpError: as :func:`check_feed_preconditions`. + """ + check_feed_preconditions(request) + feed_token = _get(request, "if_wholesale_feed_version") + if feed_token is None or feed_token != feed_version: + return False + pagination = _get(request, "pagination") + if pagination is not None and _get(pagination, "cursor") is not None: + return False + if pricing_version is not None: + return bool(_get(request, "if_pricing_version") == pricing_version) + return True + + +# ---------------------------------------------------------------------- +# Internals +# ---------------------------------------------------------------------- + + +def _get(obj: Any, name: str) -> Any: + if isinstance(obj, Mapping): + return obj.get(name) + return getattr(obj, name, None) + + +def _enum_value(value: Any) -> Any: + return getattr(value, "value", value) + + +def _mode(request: Any) -> tuple[str, Any]: + """Return the request's wholesale-mode field and value.""" + from adcp.types import GetSignalsRequest + + if isinstance(request, GetSignalsRequest) or ( + isinstance(request, Mapping) and "buying_mode" not in request + ): + # get_signals defaults discovery_mode to 'brief' when absent. + return "discovery_mode", _enum_value(_get(request, "discovery_mode")) or "brief" + return "buying_mode", _enum_value(_get(request, "buying_mode")) + + +def _wire_items(items: Iterable[Any]) -> list[Any]: + # ``_serialize`` is the products_response / signals_response item + # serializer; to_jsonable_python then settles anything a plain dict item + # left as a Python object (a datetime, an enum) to its JSON form. + return [to_jsonable_python(item) for item in _serialize(list(items))] + + +def _parse_paths(paths: Iterable[str]) -> frozenset[_Path]: + if isinstance(paths, str): + raise ValueError("ignore must be a collection of paths, not a single string") + parsed = set() + for path in paths: + parts = tuple(path.split(".")) + if not all(parts): + raise ValueError(f"invalid field path {path!r}") + parsed.add(parts) + return frozenset(parsed) + + +def _descend(paths: frozenset[_Path], key: str) -> frozenset[_Path]: + return frozenset(path[1:] for path in paths if len(path) > 1 and path[0] == key) + + +def _without(value: Any, paths: frozenset[_Path]) -> Any: + """Copy ``value`` minus the fields at ``paths``.""" + if not paths: + return value + if isinstance(value, list): + return [_without(entry, paths) for entry in value] + if isinstance(value, dict): + return { + key: _without(entry, _descend(paths, key)) + for key, entry in value.items() + if (key,) not in paths + } + return value + + +def _only(value: Any, paths: frozenset[_Path]) -> Any: + """Copy only the fields at ``paths`` out of ``value``, keeping their structure.""" + if isinstance(value, list): + return [_only(entry, paths) for entry in value] + if isinstance(value, dict): + kept: dict[str, Any] = {} + for key, entry in value.items(): + if (key,) in paths: + kept[key] = entry + elif nested := _descend(paths, key): + kept[key] = _only(entry, nested) + return kept + return None + + +def _digest(value: Any) -> str: + return hashlib.sha256(rfc8785.dumps(value)).hexdigest() + + +def _token(cache_scope: str, digests: list[str], extra: Mapping[str, Any] | None) -> str: + scope = _enum_value(cache_scope) + if scope not in _CACHE_SCOPES: + raise ValueError(f"cache_scope must be 'public' or 'account', got {cache_scope!r}") + document: dict[str, Any] = {"cache_scope": scope, "items": digests} + if extra is not None: + document["extra"] = to_jsonable_python( + { + key: _serialize(v) if isinstance(v, list) else _serialize([v])[0] + for key, v in extra.items() + } + ) + return f"sha256:{_digest(document)}" diff --git a/src/adcp/server/responses.py b/src/adcp/server/responses.py index 6d10bc1b4..dcafb1827 100644 --- a/src/adcp/server/responses.py +++ b/src/adcp/server/responses.py @@ -252,6 +252,25 @@ def _with_canonical_sources(payload: dict[str, Any], *sources: Any) -> dict[str, return _CanonicalResponse(payload, *sources) +def _check_unchanged( + unchanged: bool | None, + items_field: str, + items: list[Any] | None, + wholesale_feed_version: str | None, +) -> None: + """Reject an ``unchanged: true`` wholesale answer the schema would reject. + + ``cache_scope`` is not checked: the MCP dispatcher fills ``public`` for a + request without an account, as it does for a full response. + """ + if not unchanged: + return + if items is not None: + raise ValueError(f"an unchanged: true response must omit {items_field}") + if wholesale_feed_version is None: + raise ValueError("an unchanged: true response must echo wholesale_feed_version") + + # ============================================================================ # Protocol Discovery # ============================================================================ @@ -420,7 +439,13 @@ def products_response( metadata for cache/version-aware enumeration. Pass ``cache_scope`` explicitly for spec-valid wholesale responses; the dispatcher only infers ``public`` for request paths without an account. + + ``unchanged=True`` builds the conditional-read answer (see + :func:`adcp.server.feed_version.feed_unchanged`): ``products`` must be + omitted and ``wholesale_feed_version`` echoed, with ``cache_scope``. + ``unchanged=False`` emits nothing — the spec forbids ``unchanged: false``. """ + _check_unchanged(unchanged, "products", products, wholesale_feed_version) serialized = _serialize_canonical(products) if products is not None else None resp: dict[str, Any] = { "status": status, @@ -444,8 +469,8 @@ def products_response( resp["pricing_version"] = pricing_version if cache_scope is not None: resp["cache_scope"] = cache_scope - if unchanged is not None: - resp["unchanged"] = unchanged + if unchanged: + resp["unchanged"] = True return _with_canonical_sources(resp, products, proposals) @@ -805,8 +830,10 @@ def signals_response( Matches GetSignalsResponse schema, including beta 3 wholesale feed metadata for cache/version-aware enumeration. Pass ``cache_scope`` explicitly for spec-valid wholesale responses; the dispatcher only infers - ``public`` for request paths without an account. + ``public`` for request paths without an account. ``unchanged`` follows + :func:`products_response`. """ + _check_unchanged(unchanged, "signals", signals, wholesale_feed_version) resp: dict[str, Any] = {"status": status, "sandbox": sandbox} if signals is not None: resp["signals"] = _serialize(signals) @@ -820,8 +847,8 @@ def signals_response( resp["pricing_version"] = pricing_version if cache_scope is not None: resp["cache_scope"] = cache_scope - if unchanged is not None: - resp["unchanged"] = unchanged + if unchanged: + resp["unchanged"] = True return resp diff --git a/tests/test_feed_version.py b/tests/test_feed_version.py new file mode 100644 index 000000000..4ee463fc9 --- /dev/null +++ b/tests/test_feed_version.py @@ -0,0 +1,600 @@ +"""Seller helper for wholesale feed versions (``adcp.server.feed_version``). + +Tokens are exercised against wire dicts and real generated models; the +request rules and the ``unchanged: true`` answer are checked against the +bundled JSON schemas through the SDK's own validators, and the round trip +runs the buyer's real :class:`~adcp.feed_mirror.FeedMirror` against a seller +built from these helpers. +""" + +from __future__ import annotations + +import copy +import json +from typing import Any + +import pytest + +from adcp import FeedMirror +from adcp.decisioning import AdcpError +from adcp.server import ( + FRESHNESS_FIELDS, + PRICING_FIELDS, + check_feed_preconditions, + feed_unchanged, + pricing_version, + products_response, + signals_response, + wholesale_feed_version, +) +from adcp.types import ( + GetProductsRequest, + GetProductsResponse, + GetSignalsRequest, + GetSignalsResponse, + Product, + WholesaleFeedWebhook, +) +from adcp.types.core import TaskResult, TaskStatus +from adcp.validation.schema_validator import validate_request, validate_response +from adcp.webhooks import WebhookSender + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +def make_product(product_id: str, *, cpm: float = 18.5, name: str | None = None) -> dict[str, Any]: + return { + "product_id": product_id, + "name": name or f"Product {product_id}", + "description": f"Description for {product_id}", + "publisher_properties": [{"selection_type": "all", "publisher_domain": "pub.example.com"}], + "format_options": [ + {"format_kind": "image", "format_option_id": "image_default", "params": {}} + ], + "delivery_type": "guaranteed", + "pricing_options": [ + { + "pricing_option_id": "po_cpm", + "pricing_model": "cpm", + "currency": "USD", + "fixed_cpm": cpm, + } + ], + "reporting_capabilities": { + "available_reporting_frequencies": ["daily"], + "expected_delay_minutes": 60, + "timezone": "UTC", + "supports_webhooks": False, + "available_metrics": ["impressions"], + "date_range_support": "date_range", + }, + "forecast": { + "method": "modeled", + "currency": "USD", + "points": [{"metrics": {"impressions": {"mid": 100000}}}], + "generated_at": "2026-10-01T00:00:00Z", + "valid_until": "2026-10-02T00:00:00Z", + }, + } + + +def make_signal(segment_id: str, *, cpm: float = 2.5) -> dict[str, Any]: + return { + "signal_ref": {"scope": "product", "signal_id": segment_id}, + "signal_agent_segment_id": segment_id, + "name": f"Signal {segment_id}", + "description": f"Description for {segment_id}", + "signal_type": "marketplace", + "data_provider": "Acme Data", + "deployments": [{"type": "platform", "platform": "the-trade-desk", "is_live": True}], + "pricing_options": [ + {"pricing_option_id": "po_cpm", "model": "cpm", "cpm": cpm, "currency": "USD"} + ], + "value_type": "binary", + "coverage_forecast": { + "points": [ + { + "dimensions": [ + {"kind": "signal", "signal_id": segment_id, "presence": "present"} + ], + "metrics": {"coverage_rate": {"mid": 0.4}}, + } + ], + "forecast_range_unit": "availability", + "method": "modeled", + "scope": {"kind": "inventory", "label": "network inventory"}, + "bucket_semantics": "exclusive", + "bucket_completeness": "partial", + "generated_at": "2026-10-01T00:00:00Z", + }, + } + + +FEED = [make_product("p1"), make_product("p2", cpm=12.0), make_product("p3")] + + +def version(items: list[Any], **kwargs: Any) -> str: + kwargs.setdefault("cache_scope", "public") + return wholesale_feed_version(items, **kwargs) + + +# --------------------------------------------------------------------------- +# The token +# --------------------------------------------------------------------------- + + +def test_token_is_deterministic_and_opaque_hex() -> None: + token = version(FEED) + assert token == version(copy.deepcopy(FEED)) + assert token.startswith("sha256:") and len(token) == len("sha256:") + 64 + + +def test_model_and_wire_dict_items_give_the_same_token() -> None: + # SDK 9 writes a model's unset schema defaults (max_bid: false, ...), so a + # model hashes as its own wire form, which is what the buyer receives. + models = [Product.model_validate(item) for item in FEED] + assert version(models) == version(products_response(models)["products"]) + + +def test_top_level_order_does_not_matter() -> None: + assert version(list(reversed(FEED))) == version(FEED) + assert version([FEED[1], FEED[2], FEED[0]]) == version(FEED) + + +def test_key_order_does_not_matter() -> None: + shuffled = [dict(reversed(list(item.items()))) for item in FEED] + assert version(shuffled) == version(FEED) + + +def test_nested_array_order_is_content() -> None: + # forecast.points ascend by budget, pricing_options are presented in + # seller order: reordering inside an item is a (conservative) change. + item = make_product("p1") + item["pricing_options"].append( + {"pricing_option_id": "po_cpc", "pricing_model": "cpc", "currency": "USD", "fixed_cpc": 1} + ) + flipped = copy.deepcopy(item) + flipped["pricing_options"].reverse() + assert version([item]) != version([flipped]) + + +@pytest.mark.parametrize( + "mutate", + [ + lambda feed: feed[0].update(name="Renamed"), + lambda feed: feed[0]["pricing_options"][0].update(fixed_cpm=19.0), + lambda feed: feed[0]["forecast"]["points"][0]["metrics"]["impressions"].update(mid=1), + lambda feed: feed[0]["forecast"].update(valid_until="2026-10-03T00:00:00Z"), + lambda feed: feed.pop(), + lambda feed: feed.append(make_product("p4")), + lambda feed: feed.append(make_product("p1")), + ], + ids=["name", "price", "forecast-metric", "forecast-valid_until", "removed", "added", "dup"], +) +def test_any_buyer_visible_change_moves_the_token(mutate: Any) -> None: + changed = copy.deepcopy(FEED) + mutate(changed) + assert version(changed) != version(FEED) + + +def test_forecast_generated_at_is_ignored() -> None: + regenerated = copy.deepcopy(FEED) + for item in regenerated: + item["forecast"]["generated_at"] = "2026-10-29T09:30:00Z" + assert version(regenerated) == version(FEED) + # Opting out of the default makes it content again. + assert version(regenerated, ignore=()) != version(FEED, ignore=()) + + +def test_signal_coverage_forecast_generated_at_is_ignored() -> None: + signals = [make_signal("s1"), make_signal("s2")] + regenerated = copy.deepcopy(signals) + regenerated[0]["coverage_forecast"]["generated_at"] = "2026-10-29T09:30:00Z" + assert version(regenerated) == version(signals) + regenerated[0]["coverage_forecast"]["method"] = "measured" + assert version(regenerated) != version(signals) + + +def test_ignore_is_path_based_not_key_name_based() -> None: + # A root-level generated_at is not forecast.generated_at. + stamped = copy.deepcopy(FEED) + stamped[0]["generated_at"] = "2026-10-29T00:00:00Z" + assert version(stamped) != version(FEED) + ignore = FRESHNESS_FIELDS | {"generated_at"} + assert version(stamped, ignore=ignore) == version(FEED, ignore=ignore) + + +def test_ignore_paths_traverse_arrays() -> None: + repriced = copy.deepcopy(FEED) + repriced[0]["pricing_options"][0]["fixed_cpm"] = 99.0 + ignore = FRESHNESS_FIELDS | {"pricing_options.fixed_cpm"} + assert version(repriced, ignore=ignore) == version(FEED, ignore=ignore) + + +@pytest.mark.parametrize("ignore", ["forecast.generated_at", ["forecast..generated_at"], [""]]) +def test_malformed_ignore_paths_raise(ignore: Any) -> None: + with pytest.raises(ValueError): + version(FEED, ignore=ignore) + + +def test_cache_scope_keys_the_token() -> None: + assert version(FEED, cache_scope="public") != version(FEED, cache_scope="account") + with pytest.raises(ValueError, match="cache_scope"): + version(FEED, cache_scope="global") + + +def test_extra_response_content_moves_the_token() -> None: + # include_collections: the two settings get distinct tokens, even when + # no collection is returned. + base = version(FEED) + with_none = version(FEED, extra={"collections": []}) + with_one = version(FEED, extra={"collections": [{"collection_id": "c1", "name": "News"}]}) + assert len({base, with_none, with_one}) == 3 + + +def test_non_i_json_values_raise() -> None: + item = make_product("p1") + item["ext"] = {"big": 2**60} + with pytest.raises(ValueError): + version([item]) + + +# --------------------------------------------------------------------------- +# Separate pricing version +# --------------------------------------------------------------------------- + +STRUCTURE_ONLY = FRESHNESS_FIELDS | PRICING_FIELDS + + +def test_separated_pricing_moves_only_the_pricing_token() -> None: + repriced = copy.deepcopy(FEED) + repriced[0]["pricing_options"][0]["fixed_cpm"] = 25.0 + assert version(repriced, ignore=STRUCTURE_ONLY) == version(FEED, ignore=STRUCTURE_ONLY) + assert pricing_version(repriced, cache_scope="public") != pricing_version( + FEED, cache_scope="public" + ) + + +def test_structure_change_does_not_move_the_pricing_token() -> None: + renamed = copy.deepcopy(FEED) + renamed[0]["name"] = "Renamed" + renamed[1]["forecast"]["generated_at"] = "2026-10-29T00:00:00Z" + assert version(renamed, ignore=STRUCTURE_ONLY) != version(FEED, ignore=STRUCTURE_ONLY) + assert pricing_version(renamed, cache_scope="public") == pricing_version( + list(reversed(FEED)), cache_scope="public" + ) + + +def test_pricing_token_binds_prices_to_items() -> None: + swapped = copy.deepcopy(FEED) + swapped[0]["pricing_options"], swapped[1]["pricing_options"] = ( + swapped[1]["pricing_options"], + swapped[0]["pricing_options"], + ) + assert pricing_version(swapped, cache_scope="public") != pricing_version( + FEED, cache_scope="public" + ) + + +def test_pricing_token_covers_nested_signal_targeting_options() -> None: + def with_option(cpm: float) -> dict[str, Any]: + item = make_product("p1") + item["signal_targeting_options"] = [ + { + "signal_ref": {"scope": "product", "signal_id": "auto"}, + "name": "Auto intenders", + "pricing_options": [ + {"pricing_option_id": "s", "model": "cpm", "cpm": cpm, "currency": "USD"} + ], + } + ] + return item + + cheap, dear = [with_option(1.0)], [with_option(2.0)] + assert version(cheap, ignore=STRUCTURE_ONLY) == version(dear, ignore=STRUCTURE_ONLY) + assert pricing_version(cheap, cache_scope="public") != pricing_version( + dear, cache_scope="public" + ) + + +def test_pricing_token_covers_signals_and_needs_an_identity() -> None: + signals = [make_signal("s1"), make_signal("s2")] + repriced = [make_signal("s1", cpm=3.0), make_signal("s2")] + assert pricing_version(signals, cache_scope="account") != pricing_version( + repriced, cache_scope="account" + ) + with pytest.raises(ValueError, match="product_id or signal_agent_segment_id"): + pricing_version([{"name": "anonymous"}], cache_scope="public") + + +# --------------------------------------------------------------------------- +# Request preconditions +# --------------------------------------------------------------------------- + + +def _signals_brief(**extra: Any) -> dict[str, Any]: + return {"signal_spec": "auto intenders", **extra} + + +INVALID_REQUESTS = [ + # (tool, request, offending field) + ( + "get_products", + {"buying_mode": "wholesale", "if_pricing_version": "p"}, + "if_pricing_version", + ), + ( + "get_products", + {"buying_mode": "brief", "brief": "news", "if_wholesale_feed_version": "v"}, + "if_wholesale_feed_version", + ), + ( + "get_products", + { + "buying_mode": "refine", + "refine": [{"scope": "request", "ask": "cheaper"}], + "if_wholesale_feed_version": "v", + "if_pricing_version": "p", + }, + "if_wholesale_feed_version", + ), + ( + "get_signals", + {"discovery_mode": "wholesale", "if_pricing_version": "p"}, + "if_pricing_version", + ), + ("get_signals", _signals_brief(if_wholesale_feed_version="v"), "if_wholesale_feed_version"), +] + +VALID_REQUESTS = [ + ("get_products", {"buying_mode": "wholesale"}), + ("get_products", {"buying_mode": "wholesale", "if_wholesale_feed_version": "v"}), + ( + "get_products", + {"buying_mode": "wholesale", "if_wholesale_feed_version": "v", "if_pricing_version": "p"}, + ), + ("get_products", {"buying_mode": "brief", "brief": "news"}), + ("get_signals", {"discovery_mode": "wholesale", "if_wholesale_feed_version": "v"}), + ("get_signals", _signals_brief()), +] + +MODELS = {"get_products": GetProductsRequest, "get_signals": GetSignalsRequest} + + +@pytest.mark.parametrize("as_model", [False, True], ids=["dict", "model"]) +@pytest.mark.parametrize(("tool", "request_body", "field"), INVALID_REQUESTS) +def test_preconditions_reject_with_the_offending_field( + tool: str, request_body: dict[str, Any], field: str, as_model: bool +) -> None: + # The schema rejects the same request, so the helper and the wire agree. + assert not validate_request(tool, request_body).valid + request = MODELS[tool].model_validate(request_body) if as_model else request_body + with pytest.raises(AdcpError) as raised: + check_feed_preconditions(request) + assert raised.value.code == "INVALID_REQUEST" + assert raised.value.field == field + with pytest.raises(AdcpError): + feed_unchanged(request, feed_version="v") + + +@pytest.mark.parametrize("as_model", [False, True], ids=["dict", "model"]) +@pytest.mark.parametrize(("tool", "request_body"), VALID_REQUESTS) +def test_preconditions_accept_valid_requests( + tool: str, request_body: dict[str, Any], as_model: bool +) -> None: + assert validate_request(tool, request_body).valid + request = MODELS[tool].model_validate(request_body) if as_model else request_body + check_feed_preconditions(request) + + +# --------------------------------------------------------------------------- +# Comparator +# --------------------------------------------------------------------------- + + +def _wholesale(**tokens: Any) -> GetProductsRequest: + return GetProductsRequest(buying_mode="wholesale", **tokens) + + +def test_feed_unchanged_on_matching_token() -> None: + assert feed_unchanged(_wholesale(if_wholesale_feed_version="v1"), feed_version="v1") + assert not feed_unchanged(_wholesale(if_wholesale_feed_version="v0"), feed_version="v1") + assert not feed_unchanged(_wholesale(), feed_version="v1") + assert not feed_unchanged({"buying_mode": "brief", "brief": "news"}, feed_version="v1") + + +def test_feed_unchanged_is_false_on_continuation_pages() -> None: + request = _wholesale( + if_wholesale_feed_version="v1", pagination={"cursor": "page-2", "max_results": 10} + ) + assert not feed_unchanged(request, feed_version="v1") + first_page = _wholesale(if_wholesale_feed_version="v1", pagination={"max_results": 10}) + assert feed_unchanged(first_page, feed_version="v1") + + +def test_feed_unchanged_pricing_evaluation_order() -> None: + both = _wholesale(if_wholesale_feed_version="v1", if_pricing_version="p1") + assert feed_unchanged(both, feed_version="v1", pricing_version="p1") + # (2) feed matches, pricing does not -> full payload. + assert not feed_unchanged(both, feed_version="v1", pricing_version="p2") + # (1) feed mismatch -> full payload regardless of pricing. + assert not feed_unchanged(both, feed_version="v2", pricing_version="p1") + # A seller that separates pricing cannot vouch for prices from the feed + # token alone. + feed_only = _wholesale(if_wholesale_feed_version="v1") + assert not feed_unchanged(feed_only, feed_version="v1", pricing_version="p1") + # A seller that does not track pricing ignores if_pricing_version. + assert feed_unchanged(both, feed_version="v1") + + +# --------------------------------------------------------------------------- +# The unchanged answer +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("tool", "builder"), [("get_products", products_response), ("get_signals", signals_response)] +) +def test_unchanged_answer_validates_against_the_schema(tool: str, builder: Any) -> None: + body = builder( + unchanged=True, + wholesale_feed_version=version(FEED), + pricing_version=pricing_version(FEED, cache_scope="public"), + cache_scope="public", + ) + assert body["unchanged"] is True + assert "products" not in body and "signals" not in body + outcome = validate_response(tool, body) + assert outcome.valid, outcome.issues + + +def test_unchanged_false_is_never_emitted() -> None: + body = products_response( + FEED, wholesale_feed_version=version(FEED), cache_scope="public", unchanged=False + ) + assert "unchanged" not in body + assert validate_response("get_products", body).valid + signals = [make_signal("s1")] + body = signals_response( + signals, wholesale_feed_version=version(signals), cache_scope="public", unchanged=False + ) + assert "unchanged" not in body + assert validate_response("get_signals", body).valid + + +@pytest.mark.parametrize("builder", [products_response, signals_response]) +def test_malformed_unchanged_answers_raise(builder: Any) -> None: + with pytest.raises(ValueError, match="must omit"): + builder([], unchanged=True, wholesale_feed_version="v", cache_scope="public") + with pytest.raises(ValueError, match="must echo"): + builder(unchanged=True, cache_scope="public") + # cache_scope may be left to the dispatcher, which fills "public" for a + # request without an account. + assert builder(unchanged=True, wholesale_feed_version="v")["unchanged"] is True + + +# --------------------------------------------------------------------------- +# Round trip: buyer FeedMirror against a seller built from these helpers +# --------------------------------------------------------------------------- + + +class HelperSeller: + """A wholesale seller that answers through the helpers and the schema.""" + + def __init__(self, products: list[dict[str, Any]], *, separate_pricing: bool = False) -> None: + self.products = products + self.separate_pricing = separate_pricing + self.responses: list[dict[str, Any]] = [] + + def tokens(self) -> tuple[str, str | None]: + if not self.separate_pricing: + return version(self.products), None + return ( + version(self.products, ignore=STRUCTURE_ONLY), + pricing_version(self.products, cache_scope="public"), + ) + + async def get_products(self, request: GetProductsRequest) -> TaskResult[GetProductsResponse]: + assert validate_request("get_products", request).valid + feed_token, price_token = self.tokens() + if feed_unchanged(request, feed_version=feed_token, pricing_version=price_token): + body = products_response( + unchanged=True, + wholesale_feed_version=feed_token, + pricing_version=price_token, + cache_scope="public", + ) + else: + # Two-page walk: every page carries the whole feed's token. + cursor = request.pagination.cursor if request.pagination else None + start = int(cursor) if cursor else 0 + page = self.products[start : start + 2] + more = start + 2 < len(self.products) + body = products_response( + page, + wholesale_feed_version=feed_token, + pricing_version=price_token, + cache_scope="public", + pagination={"has_more": more, **({"cursor": str(start + 2)} if more else {})}, + ) + outcome = validate_response("get_products", body) + assert outcome.valid, outcome.issues + self.responses.append(body) + return TaskResult( + status=TaskStatus.COMPLETED, + success=True, + data=GetProductsResponse.model_validate(json.loads(json.dumps(body))), + ) + + async def get_signals(self, request: GetSignalsRequest) -> TaskResult[GetSignalsResponse]: + raise AssertionError("products only") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("separate_pricing", [False, True], ids=["single", "separate-pricing"]) +async def test_feed_mirror_round_trip_honours_unchanged(separate_pricing: bool) -> None: + seller = HelperSeller(copy.deepcopy(FEED), separate_pricing=separate_pricing) + mirror = FeedMirror(seller, strict_scope=True) + + await mirror.bootstrap("product") + assert set(mirror.products) == {"p1", "p2", "p3"} + assert len(seller.responses) == 2 # paginated full load + + # Same content, regenerated forecast: the mirror short-circuits. + for product in seller.products: + product["forecast"]["generated_at"] = "2026-10-29T09:30:00Z" + result = await mirror.refresh("product") + assert result.products_unchanged + assert seller.responses[-1]["unchanged"] is True + assert set(mirror.products) == {"p1", "p2", "p3"} + + # A re-price is a full payload, with or without a separate pricing token. + seller.products[0]["pricing_options"][0]["fixed_cpm"] = 30.0 + result = await mirror.refresh("product") + assert not result.products_unchanged + assert "unchanged" not in seller.responses[-1] + price = mirror.products["p1"].pricing_options[0].fixed_cpm + assert price == 30.0 + + result = await mirror.refresh("product") + assert result.products_unchanged + + +@pytest.mark.asyncio +async def test_webhook_version_matches_the_conditional_read_token() -> None: + seller = HelperSeller(copy.deepcopy(FEED)) + mirror = FeedMirror(seller, strict_scope=True) + await mirror.bootstrap("product") + previous = mirror.product_state.wholesale_feed_version + + # The seller removes p3 and announces it with the post-change token. + seller.products = [p for p in seller.products if p["product_id"] != "p3"] + current = version(seller.products) + prepared = WebhookSender.prepare_wholesale_feed( + url="https://buyer.example/webhooks/feed", + subscriber_id="buyer-primary", + account_id="acct_1", + notification_type="product.removed", + wholesale_feed_version=current, + previous_wholesale_feed_version=previous, + cache_scope="public", + event={ + "event_id": "018f13f8-7b40-7000-8000-000000000123", + "event_type": "product.removed", + "entity_type": "product", + "entity_id": "p3", + "created_at": "2026-10-29T12:00:00Z", + "payload": {"product_id": "p3", "applies_to": {"scope": "public"}}, + }, + ) + webhook = WholesaleFeedWebhook.model_validate_json(prepared.body) + assert webhook.wholesale_feed_version == current + await mirror.apply_webhook(webhook) + + # The buyer presents the webhook's version; the seller recognises it. + result = await mirror.refresh("product") + assert result.products_unchanged + assert seller.responses[-1]["wholesale_feed_version"] == current + assert set(mirror.products) == {"p1", "p2"}