From f119e57e0a53aaecf26e9a04edb2c4f58ceaa492 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 00:17:36 +0000 Subject: [PATCH] feat(accounts)!: enforce change feed changed_paths and per-reader idempotency_key (#1647) AdCP 3.3 account change feed rules, applied where the SDK emits records: - list_account_changes dispatch refuses a page carrying an adcp / seller_operator record without non-empty changed_paths (VALIDATION_ERROR at /changes/N/changed_paths, any validation mode, handlers on 3.3 or later). - Every record loses idempotency_key unless it is an adcp record attributed, off the wire, to the calling principal (attribute_account_change); unattributed records fail closed. - WebhookSender.prepare_account_change_recorded accepts the change record (with subscriber_id), checks it, and never carries its key. - New adcp.account_change_feed helpers: json_pointer, ROOT_POINTER, validate_account_change, attribute_account_change, project_list_account_changes_response, account_change_recorded_payload, mirror_creative_revision (creative seller_revision -> resource_revision). BREAKING CHANGE: a list_account_changes handler returning an adcp or seller_operator record without changed_paths now fails the call, and idempotency_key is stripped from every record not attributed to the caller via adcp.account_change_feed.attribute_account_change. Co-Authored-By: Claude Opus 5.5 --- MIGRATION_v9_to_v10.md | 1 + docs/adcp-3.3-sdk-inventory.md | 4 +- src/adcp/account_change_feed.py | 385 ++++++++++++++++++++++++++ src/adcp/server/mcp_tools.py | 54 ++++ src/adcp/webhook_sender.py | 60 ++++- tests/test_account_change_feed.py | 432 ++++++++++++++++++++++++++++++ 6 files changed, 931 insertions(+), 5 deletions(-) create mode 100644 src/adcp/account_change_feed.py create mode 100644 tests/test_account_change_feed.py diff --git a/MIGRATION_v9_to_v10.md b/MIGRATION_v9_to_v10.md index 74192bfa39..0fe0179dc9 100644 --- a/MIGRATION_v9_to_v10.md +++ b/MIGRATION_v9_to_v10.md @@ -48,3 +48,4 @@ the SDK-relevant inventory is [docs/adcp-3.3-sdk-inventory.md](docs/adcp-3.3-sdk | Unknown keys on open objects are kept and written back | `AdCPBaseModel` defaults to `extra='allow'`: a property the pinned model does not declare (a field a newer spec added, a sibling arm's property the schema admits) survives `model_validate` → `model_dump` in `model_extra`, including a `null` value. Generated classes whose schema object is closed (`additionalProperties: false`) keep `extra='forbid'`. SDK 9 dropped unknown keys (`extra='ignore'`). If your code passed stray keyword arguments and relied on them being dropped, remove them; `ADCP_STRICT_VALIDATION=1` turns the default into `forbid` so tests find them. | | Permissive adagents.json resolution requires the fetched domain | `resolve_properties_for_agent(doc, agent_url, mode="permissive")` raises `ValueError` without `publisher_domain=`. Pass the domain the document was fetched for: a bare entry's top-level properties are then kept only when every identifier is a `type: "domain"` under that domain, so a file on another host can no longer claim a publisher's site, app bundle or DOOH venue through a bare entry. Properties identified by bundle, store or venue IDs, and properties whose `publisher_domain` names another domain, are dropped from the bare fallback. `allow_any_domain=True` restores the SDK 9 result. Typed selectors and strict mode are unchanged. See [docs/adagents-discovery.md](docs/adagents-discovery.md) for `properties_for_publisher_domain` and `verify_cross_domain_properties`. | | Generated enums name each value once | The `_1` alias members codegen emitted when an `anyOf` arm repeated an enum value (`System.zip_1`, `Field17.ad_request_url_1`, 19 enums on 10.0.0b4) are gone; each value has one member, the first-declared one. Code that referenced an alias name (`System.zip_1`) must use the real member (`System.zip`); `System("zip")` and `member.value` are unchanged. | +| `list_account_changes` pages are checked and projected per caller | AdCP 3.3 requires non-empty `changed_paths` on every `adcp` / `seller_operator` change record (`[""]` for a whole-resource create, delete or purge) and lets the originating request's `idempotency_key` reach only the principal that sent it. A handler on 3.3 or later that returns such a record without `changed_paths` now fails the call with `VALIDATION_ERROR` (`field` `/changes/N/changed_paths`), whatever the validation mode; and every record loses `idempotency_key` unless it is an `adcp` record built with `adcp.account_change_feed.attribute_account_change(record, originator)` whose originator is the caller (`ToolContext.tenant_id` + `caller_identity`). Persist the originator when the mutation commits and attribute records when serving the page. `WebhookSender.prepare_account_change_recorded` / `send_account_change_recorded` also accept the change record itself (with `subscriber_id=`), refuse one without `changed_paths`, and never carry its `idempotency_key`. | diff --git a/docs/adcp-3.3-sdk-inventory.md b/docs/adcp-3.3-sdk-inventory.md index 126300e742..2dcd5f53ca 100644 --- a/docs/adcp-3.3-sdk-inventory.md +++ b/docs/adcp-3.3-sdk-inventory.md @@ -63,7 +63,7 @@ it adopted the bundle, and what is left to the codegen/wire-fidelity (2) and | `creative_assignments_as_of`, `approval_status_as_of` | `get_media_buys` `Package`, `CreativeApproval`; `list_creatives` `AssignedPackage` | Timestamps. | generated | | `landing_page_url`, `trackers` on assignments | `core/creative-assignment.json`, `core/assignment-tracker.json`; `sync_creatives` `Assignment` / `AssignmentOperations1`; `list_creatives` `AssignedPackage`; `get_media_buys` `CreativeApproval`; capabilities `creative.supports_assignment_landing_page`, `creative.supports_assignment_trackers` | Per-assignment overrides. | generated | | `sync_creatives` `merges[]`, `AssignmentOperations4` | `sync_creatives` response `Merge`; error `CREATIVE_MERGE_REJECTED` with `error-details/creative-merge-rejected.json`; enum `creative-merge-rejection-reason`; capability `creative.supports_assignment_merge` | Creative merge. | generated; `CreativeMergeRejectedError` mints schema-valid details (field defaults to the first rejected `assignment_operations` entry) and `creative_merge_rejected_details` narrows them: done (3), #1668 | -| `list_creatives` `platform_id`, `provenance`, `seller_revision`, `status_as_of`; request filter `provenance` | `creative/list-creatives-response.json`; `x-entity-types` `creative_seller_revision`; capability `creative.stable_creative_ids` | `platform_id` is native in 3.3 (adcp#8175). There was no forward-compat patch for it in this SDK; nothing to retire. | generated; `stable_creative_ids` seller support and `delete_missing` exclusion: (3), #1638 | +| `list_creatives` `platform_id`, `provenance`, `seller_revision`, `status_as_of`; request filter `provenance` | `creative/list-creatives-response.json`; `x-entity-types` `creative_seller_revision`; capability `creative.stable_creative_ids` | `platform_id` is native in 3.3 (adcp#8175). There was no forward-compat patch for it in this SDK; nothing to retire. | generated; `seller_revision` passes through `list_creatives` unharmed and is dropped (loss recorded) when served to a buyer below 3.3; `adcp.account_change_feed.mirror_creative_revision` copies it into a creative change record's `resource_revision` in the same JSON type: done (3), #1647; `stable_creative_ids` seller support and `delete_missing` exclusion: (3), #1638 | | `validate_input` `Warning.field` | `creative/validate-input-result.json` | `field` is now declared as a string (#8212). | generated | | `build_creative` `errors` | `BuildCreativeResponse1/3` | Errors array on build responses. | generated | | `pixel_tracker` events | enum `pixel-tracking-event`: `start`, `first_quartile`, `midpoint`, `third_quartile`, `complete` | Send only to 3.3 peers. Interim form for 3.2 peers: `event: "custom"` with `custom_event_name`; the 3.3 → 3.2 downgrade may rewrite to that form. Downgrade to 3.0 `tracker_pixel` drops them with `PIXEL_TRACKER_LOSSY_DOWNGRADE`, never as `impression_tracker`. | generated; 3.0 downgrade drop (including the `custom` interim form): done (3), #1655; 3.3 → 3.2 rewrite to `custom` in `convert_payload`, buyer requests and seller responses (loss `playback_event_as_custom`): done (3), #1643 | @@ -109,7 +109,7 @@ it adopted the bundle, and what is left to the codegen/wire-fidelity (2) and | --- | --- | --- | --- | | `account.stable_account_id`, `account_id` from `sync_accounts` | capabilities `Account.stable_account_id` | Natural-key references after provisioning are deprecated. | seller `NaturalKeyAccounts(loader_by_id=...)`: done (3), #1653; capability declaration and buyer `AccountRegistry.prepare`: (3), #1478 | | Discovery never provisions | `get_products`, `list_products`, `get_signals`, `request_proposals`, `refine_proposals`, `decline_proposals` MUST NOT create or activate an account; unresolved `account` → `ACCOUNT_NOT_FOUND`; `list_accounts` treats `account` as a filter | Storyboard `media_buy_seller/unprovisioned_account_reference` (advisory until runner 14.1.0). | done: discovery tasks never provision (`ResolveContext.is_provisioning`, #1288) and a supplied unresolved reference is `ACCOUNT_NOT_FOUND`; confirm with the storyboard: (3) | -| `account-change.json` `idempotency_key`; `changed_paths` required for `adcp` / `seller_operator` origins | `AccountChange.idempotency_key` | Echoed to the sending principal only. | generated; account change feed: (3), #1647 | +| `account-change.json` `idempotency_key`; `changed_paths` required for `adcp` / `seller_operator` origins | `AccountChange.idempotency_key` | Echoed to the sending principal only. | generated; `list_account_changes` dispatch refuses a page with an `adcp` / `seller_operator` record lacking non-empty `changed_paths` (`VALIDATION_ERROR` at `/changes/N/changed_paths`, any validation mode, handlers on 3.3 or later) and strips `idempotency_key` from every record except an `adcp` record attributed (`attribute_account_change`, off the wire) to the calling principal; `WebhookSender.prepare_account_change_recorded` accepts the record, checks it, and never carries its key; helpers in `adcp.account_change_feed`: done (3), #1647 | | `account.agent_onboarding`, `get_principal` `relationship`, `identity.trust_json`, `identity-relying-party` | `core/agent-onboarding.json`, `-tier.json`, `-requirement.json`, `core/buyer-onboarding-posture.json`, `core/principal-relationship.json`, `trust/v1/trust.json`, `trust/v1/trust-acknowledgements.json`; enums `agent-onboarding-mode`, `agent-onboarding-requirement-kind` (x-extensible), `principal-relationship-status`; `PrincipalChangedWebhook.Reason` `relationship_changed`, `relationship_expiring`; `get_principal` `Result.errors` | Experimental. The identity preview (`_identity_preview.py`, vendored `_identity_schemas/`, `generate_identity_preview.py`) is retired: `adcp.types.experimental_identity` binds the generated 3.3 types, `AgentOnboardingRequirement.kind` stays open (`enum | str`), and `adcp.signing.trust` validates against the bundle through `get_named_validator`. | done (1) | | `adcp.implementation.sdks[]` | capabilities `Adcp.implementation` (`Implementation`, `Sdk`, `Component`) | Vendors move data out of `ext.adcp`. | generated; advertise this SDK: (3), #1646 | | `security_profiles` | `GetAdcpCapabilitiesResponse.security_profiles` (`adcp-prod-security-3.3`, experimental) | | generated | diff --git a/src/adcp/account_change_feed.py b/src/adcp/account_change_feed.py new file mode 100644 index 0000000000..60f0f78cef --- /dev/null +++ b/src/adcp/account_change_feed.py @@ -0,0 +1,385 @@ +"""Account change feed records (AdCP 3.3, experimental ``account.change_feed``). + +A seller that declares the change feed serves ``list_account_changes`` and +may push ``account.change_recorded`` notifications. Three record rules apply, +and this module is where the SDK checks them: + +* **changed_paths.** A record whose ``origin.kind`` is ``adcp`` or + ``seller_operator`` carries a non-empty ``changed_paths``: RFC 6901 pointers + naming the material fields that changed, or the root pointer ``""`` + (:data:`ROOT_POINTER`) for a whole-resource create, delete or purge. The + seller processed those mutations, so it knows what it changed. +* **idempotency_key.** An ``adcp`` record may echo the ``idempotency_key`` of + the request that produced it, but only to the principal that sent that + request. Every other reader of the account gets the record without it, and + ``account.change_recorded`` notifications never carry it. +* **resource_revision.** For a creative, ``resource_revision`` mirrors the + creative's ``seller_revision`` after the change, in the same JSON type. + +The originating principal is not on the wire, so the seller attaches it to the +record with :func:`attribute_account_change` when it builds the page. The +server's ``list_account_changes`` dispatch then projects every record for the +caller: an attributed ``adcp`` record keeps its key only when the caller is +its originator, and any other record loses it. Unattributed records always +lose it, so the projection fails closed:: + + from adcp.account_change_feed import ROOT_POINTER, attribute_account_change + + # When the mutation commits, persist the originator with the record: + originator = principal_ref_for(ctx) # ctx: the mutating request's ToolContext + + # When serving list_account_changes: + changes = [attribute_account_change(row.record, row.originator) for row in page] + return ListAccountChangesResponse(status="completed", changes=changes, ...) + +Build notifications from the same record with +:func:`account_change_recorded_payload`, or pass the record straight to +:meth:`adcp.webhook_sender.WebhookSender.prepare_account_change_recorded`. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any, Final, TypeVar + +from pydantic import BaseModel, PrivateAttr + +from adcp._principal_ref import PrincipalRef +from adcp.types.domains.core.account_change import AccountChange + +__all__ = [ + "ROOT_POINTER", + "AccountChangeRecordError", + "AttributedAccountChange", + "account_change_recorded_payload", + "attribute_account_change", + "is_account_change_record", + "json_pointer", + "mirror_creative_revision", + "principal_ref_for", + "project_account_change", + "project_list_account_changes_response", + "validate_account_change", +] + +#: RFC 6901 root pointer: the whole resource was created, deleted or purged. +ROOT_POINTER: Final = "" + +#: Origins whose records must list what they changed. +_PATHS_REQUIRED_ORIGINS: Final = frozenset({"adcp", "seller_operator"}) + +_RECORD_ONLY_KEYS: Final = frozenset({"origin", "repair"}) + +_ChangeT = TypeVar("_ChangeT", bound="AccountChange | Mapping[str, Any]") + + +class AccountChangeRecordError(ValueError): + """An account change record breaks an AdCP 3.3 change-feed rule. + + ``pointer`` is the JSON Pointer of the offending field, relative to the + payload the record was checked in (``/changes/3/changed_paths`` in a + ``list_account_changes`` response). + """ + + def __init__(self, message: str, *, pointer: str = "") -> None: + super().__init__(message) + self.pointer = pointer + + +class AttributedAccountChange(AccountChange): + """An :class:`AccountChange` that knows, off the wire, who originated it. + + The originating principal is a private attribute: it never serializes and + does not affect validation. Build one with :func:`attribute_account_change`. + """ + + _originating_principal: PrincipalRef | None = PrivateAttr(default=None) + + @property + def originating_principal(self) -> PrincipalRef | None: + """The principal that sent the AdCP request behind this change, if known.""" + return self._originating_principal + + +def json_pointer(*tokens: str | int) -> str: + """Build an RFC 6901 JSON Pointer from unescaped reference tokens. + + ``json_pointer("packages", 0, "budget")`` is ``"/packages/0/budget"``; + ``~`` and ``/`` inside a token are escaped. With no tokens the result is + the root pointer ``""``. + """ + return "".join("/" + str(token).replace("~", "~0").replace("/", "~1") for token in tokens) + + +def principal_ref_for( + principal: PrincipalRef | str | Any | None, *, tenant_id: str | None = None +) -> PrincipalRef | None: + """The :class:`~adcp.PrincipalRef` of a caller, or ``None`` when it is unknown. + + Accepts a ``PrincipalRef``, a principal id string (scoped to ``tenant_id``) + or a :class:`~adcp.server.ToolContext`-shaped object exposing + ``caller_identity`` and ``tenant_id``. A context without an authenticated + caller yields ``None``. + """ + if principal is None or isinstance(principal, PrincipalRef): + return principal + if isinstance(principal, str): + return PrincipalRef(tenant_id or None, principal) if principal else None + if hasattr(principal, "caller_identity"): + caller = principal.caller_identity + if not isinstance(caller, str) or not caller: + return None + tenant = getattr(principal, "tenant_id", None) + return PrincipalRef(tenant if isinstance(tenant, str) and tenant else None, caller) + return PrincipalRef.coerce(principal, tenant_id=tenant_id) + + +def _field(record: Any, name: str) -> Any: + if isinstance(record, Mapping): + return record.get(name) + return getattr(record, name, None) + + +def _origin_kind(record: Any) -> str | None: + origin = _field(record, "origin") + if origin is None: + return None + kind = _field(origin, "kind") + if kind is None: + return None + return str(getattr(kind, "value", kind)) + + +def _describe(record: Any) -> str: + change_id = _field(record, "change_id") + return f"account change {change_id!r}" if isinstance(change_id, str) else "account change" + + +def _require_changed_paths(change: Any, *, pointer: str) -> None: + kind = _origin_kind(change) + if kind in _PATHS_REQUIRED_ORIGINS and not _field(change, "changed_paths"): + raise AccountChangeRecordError( + f"{_describe(change)} has origin.kind {kind!r} but no changed_paths: list the " + 'RFC 6901 pointers of the fields it changed, or [""] for a whole-resource ' + "create, delete or purge", + pointer=f"{pointer}/changed_paths", + ) + + +def validate_account_change( + change: AccountChange | Mapping[str, Any], *, pointer: str = "" +) -> None: + """Check a record against the change-feed rules the generated model does not enforce. + + Raises :class:`AccountChangeRecordError` when an ``adcp`` or + ``seller_operator`` record has no non-empty ``changed_paths``, or when a + record that is not ``adcp``-originated carries an ``idempotency_key``. + ``pointer`` locates the record in its payload for the error. + """ + _require_changed_paths(change, pointer=pointer) + kind = _origin_kind(change) + if kind != "adcp" and _field(change, "idempotency_key") is not None: + raise AccountChangeRecordError( + f"{_describe(change)} carries idempotency_key but origin.kind is {kind!r}; " + "only adcp-originated records echo the originating request's key", + pointer=f"{pointer}/idempotency_key", + ) + + +def attribute_account_change( + change: AccountChange | Mapping[str, Any], + originating_principal: PrincipalRef | str | Any | None, + *, + tenant_id: str | None = None, +) -> AttributedAccountChange: + """Validate a change record and attach the principal that originated it. + + ``originating_principal`` is the principal that sent the AdCP request + behind an ``adcp`` change: a ``PrincipalRef``, a principal id (scoped to + ``tenant_id``), or the mutating request's ``ToolContext``. Pass ``None`` + for other origins, or when the originator is unknown; the record's + ``idempotency_key`` is then withheld from every reader. + + Raises :class:`AccountChangeRecordError` for a record that breaks the + ``changed_paths`` or ``idempotency_key`` rules, and + :class:`pydantic.ValidationError` for one that fails the schema model. + """ + data = ( + change.model_dump(mode="json", exclude_none=True) + if isinstance(change, BaseModel) + else dict(change) + ) + record = AttributedAccountChange.model_validate(data) + validate_account_change(record) + record._originating_principal = principal_ref_for(originating_principal, tenant_id=tenant_id) + return record + + +def _echoes_key_to(change: Any, reader: PrincipalRef | None) -> bool: + if reader is None or _origin_kind(change) != "adcp": + return False + if not isinstance(change, AttributedAccountChange): + return False + originator = change.originating_principal + return originator is not None and originator == reader + + +def project_account_change( + change: AccountChange | Mapping[str, Any], *, reader: PrincipalRef | str | Any | None +) -> dict[str, Any]: + """The wire form of one record as served to ``reader``. + + ``idempotency_key`` survives only on an :class:`AttributedAccountChange` + with ``origin.kind == "adcp"`` whose originating principal equals + ``reader``. Every other record, including a plain mapping or an + unattributed model, is returned without it. + """ + if isinstance(change, BaseModel): + data = change.model_dump(mode="json", exclude_none=True) + elif isinstance(change, Mapping): + data = dict(change) + else: + raise TypeError(f"expected an account change record, got {type(change).__name__}") + if "idempotency_key" in data and not _echoes_key_to(change, principal_ref_for(reader)): + del data["idempotency_key"] + return data + + +def project_list_account_changes_response( + result: Any, + *, + reader: PrincipalRef | str | Any | None, + require_changed_paths: bool = True, +) -> Any: + """Project a ``list_account_changes`` result for the caller ``reader``. + + Each record goes through :func:`project_account_change`, which also drops + a key a non-``adcp`` record should not carry. With + ``require_changed_paths`` (AdCP 3.3 and later), an ``adcp`` / + ``seller_operator`` record without non-empty ``changed_paths`` raises + :class:`AccountChangeRecordError`, so a non-conformant page is refused + instead of shipping. A model + result comes back as a JSON-mode dict; a result without ``changes`` + (an error envelope) comes back unchanged. + """ + reader_ref = principal_ref_for(reader) + if isinstance(result, BaseModel): + records = getattr(result, "changes", None) + if records is None: + return result + data = result.model_dump(mode="json", exclude_none=True, exclude={"changes"}) + elif isinstance(result, Mapping): + records = result.get("changes") + if not isinstance(records, (list, tuple)): + return result + data = dict(result) + else: + return result + projected = [] + for index, record in enumerate(records): + if require_changed_paths: + _require_changed_paths(record, pointer=f"/changes/{index}") + projected.append(project_account_change(record, reader=reader_ref)) + data["changes"] = projected + return data + + +def is_account_change_record(payload: Any) -> bool: + """Whether ``payload`` is a change record rather than a notification body. + + ``origin`` and ``repair`` are required on a record and forbidden on an + ``account.change_recorded`` notification. + """ + if isinstance(payload, AccountChange): + return True + return isinstance(payload, Mapping) and not _RECORD_ONLY_KEYS.isdisjoint(payload) + + +def account_change_recorded_payload( + change: AccountChange | Mapping[str, Any], + *, + subscriber_id: str, + through_cursor: str | None = None, + ext: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Build the ``account.change_recorded`` wire body for a change record. + + Copies the record's identity (``change_id`` as ``notification_id``, + ``resource.account_id`` as ``account_id``, ``recorded_at``, ``resource`` + and ``action``). The record's ``idempotency_key`` is never copied: it + belongs to the originating request, and notifications MUST NOT carry it. + The delivery ``idempotency_key`` and ``fired_at`` are filled at + preparation. Raises :class:`AccountChangeRecordError` for a record that + breaks the change-feed rules. + """ + record = change if isinstance(change, AccountChange) else AccountChange.model_validate(change) + validate_account_change(record) + wire = record.model_dump(mode="json", exclude_none=True) + resource = wire["resource"] + payload: dict[str, Any] = { + "notification_type": "account.change_recorded", + "notification_id": wire["change_id"], + "subscriber_id": subscriber_id, + "account_id": resource["account_id"], + "change_id": wire["change_id"], + "recorded_at": wire["recorded_at"], + "resource": { + key: resource[key] for key in ("type", "resource_id", "parent_ids") if key in resource + }, + "action": wire["action"], + } + if through_cursor is not None: + payload["through_cursor"] = through_cursor + if ext is not None: + payload["ext"] = dict(ext) + return payload + + +def _seller_revision(creative: Any) -> int | str | None: + revision = _field(creative, "seller_revision") + if revision is None: + return None + if isinstance(revision, bool) or not isinstance(revision, (int, str)): + raise AccountChangeRecordError( + "creative seller_revision must be a positive integer or a 1-255 character string", + pointer="/resource_revision", + ) + if isinstance(revision, int) and revision < 1: + raise AccountChangeRecordError( + "creative seller_revision must be a positive integer", pointer="/resource_revision" + ) + if isinstance(revision, str) and not 1 <= len(revision) <= 255: + raise AccountChangeRecordError( + "creative seller_revision must be 1 to 255 characters", pointer="/resource_revision" + ) + # A generated scalar subclass dumps as its base type; keep the JSON type. + return int(revision) if isinstance(revision, int) else str(revision) + + +def mirror_creative_revision(change: _ChangeT, creative: Any) -> _ChangeT: + """Return ``change`` with ``resource_revision`` mirroring the creative's ``seller_revision``. + + ``creative`` is the creative as the repair read (``list_creatives``) + reports it after the change, as a model or a mapping. Its + ``seller_revision`` is copied in the same JSON type; when the creative has + none, ``resource_revision`` is removed. The input is not modified, and an + :class:`AttributedAccountChange` keeps its originator. + + Raises :class:`AccountChangeRecordError` when the record's resource is not + a creative. + """ + resource_type = _field(_field(change, "resource"), "type") + if resource_type != "creative": + raise AccountChangeRecordError( + f"{_describe(change)} is for resource type {resource_type!r}, not creative", + pointer="/resource/type", + ) + revision = _seller_revision(creative) + if isinstance(change, AccountChange): + return change.model_copy(update={"resource_revision": revision}) + updated = dict(change) + if revision is None: + updated.pop("resource_revision", None) + else: + updated["resource_revision"] = revision + return updated # type: ignore[return-value] diff --git a/src/adcp/server/mcp_tools.py b/src/adcp/server/mcp_tools.py index d8f60d595a..27040fdb77 100644 --- a/src/adcp/server/mcp_tools.py +++ b/src/adcp/server/mcp_tools.py @@ -3344,6 +3344,58 @@ def _apply_unknown_field_policy( ) +def _requires_changed_paths(application_version: str | None) -> bool: + """Whether the handler's release requires ``changed_paths`` (AdCP 3.3 and later).""" + from adcp._version import normalize_to_release_precision + + if application_version is None: + return True + try: + release = normalize_to_release_precision(application_version) + major, minor = release.split("-", 1)[0].split(".")[:2] + return (int(major), int(minor)) >= (3, 3) + except (TypeError, ValueError): + return True + + +def _project_account_change_feed( + result: Any, ctx: ToolContext, application_version: str | None +) -> Any: + """Project a ``list_account_changes`` page for the caller (#1647). + + Records lose ``idempotency_key`` unless the caller originated them, and a + page carrying an ``adcp`` / ``seller_operator`` record without + ``changed_paths`` is refused rather than served. + """ + from adcp.account_change_feed import ( + AccountChangeRecordError, + project_list_account_changes_response, + ) + + try: + return project_list_account_changes_response( + result, + reader=ctx, + require_changed_paths=_requires_changed_paths(application_version), + ) + except AccountChangeRecordError as exc: + from adcp.exceptions import ADCPTaskError + from adcp.types import Error + + logger.error("list_account_changes handler returned a non-conformant record: %s", exc) + raise ADCPTaskError( + operation="list_account_changes", + errors=[ + Error( + code="VALIDATION_ERROR", + field=exc.pointer or None, + message=f"list_account_changes response failed validation: {exc}", + details={"tool": "list_account_changes", "side": "response"}, + ) + ], + ) from exc + + def _creative_capabilities_for_handler(handler: Any) -> Any: """Return deterministic server-owned creative capability evidence. @@ -4105,6 +4157,8 @@ async def call_tool(params: dict[str, Any], context: ToolContext | None = None) ], ) from exc result = await method(call_params, ctx) + if method_name == "list_account_changes": + result = _project_account_change_feed(result, ctx, application_version) if method_name == "get_adcp_capabilities": if isinstance(result, dict): from adcp.canonical_formats.dialect import canonical_creatives_capability diff --git a/src/adcp/webhook_sender.py b/src/adcp/webhook_sender.py index 2e3f68d8bb..b4db0292ce 100644 --- a/src/adcp/webhook_sender.py +++ b/src/adcp/webhook_sender.py @@ -979,12 +979,58 @@ def prepare_account_change_recorded( payload: Mapping[str, Any] | BaseModel, idempotency_key: str | None = None, extra_headers: Mapping[str, str] | None = None, + subscriber_id: str | None = None, + through_cursor: str | None = None, ) -> PreparedWebhook: """Prepare a schema-validated ``account.change_recorded`` notification. - Supply the wire fields as a mapping or a Pydantic model. The type, - fire timestamp and delivery key are filled when absent. + ``payload`` is either the notification's wire fields (a mapping or a + Pydantic model) or the :class:`~adcp.types.domains.core.account_change.AccountChange` + record it announces. The type, fire timestamp and delivery key are + filled when absent. + + From a record, the body is built by + :func:`adcp.account_change_feed.account_change_recorded_payload`: + ``subscriber_id`` is required and ``through_cursor`` optional, a record + that breaks the change-feed rules (an ``adcp`` / ``seller_operator`` + origin without ``changed_paths``) raises + :class:`~adcp.account_change_feed.AccountChangeRecordError`, and the + record's ``idempotency_key`` is never carried: the delivery key is + ``idempotency_key`` when given, else freshly minted, and may not equal + the originating request's key. Notification bodies are passed through; + their ``idempotency_key`` is the delivery key, so never copy a record's + key into one. """ + from adcp.account_change_feed import ( + account_change_recorded_payload, + is_account_change_record, + ) + + if is_account_change_record(payload): + if not subscriber_id: + raise ValueError( + "preparing account.change_recorded from a change record needs subscriber_id" + ) + source_key = ( + payload.get("idempotency_key") + if isinstance(payload, Mapping) + else getattr(payload, "idempotency_key", None) + ) + if source_key is not None and idempotency_key == source_key: + raise ValueError( + "account.change_recorded delivery idempotency_key must not reuse the " + "originating request's idempotency_key" + ) + payload = account_change_recorded_payload( + payload, # type: ignore[arg-type] + subscriber_id=subscriber_id, + through_cursor=through_cursor, + ) + elif subscriber_id is not None or through_cursor is not None: + raise ValueError( + "subscriber_id and through_cursor apply only when payload is a change record; " + "set them in the notification payload" + ) return WebhookSender._prepare_notification( url=url, notification_type="account.change_recorded", @@ -1000,14 +1046,22 @@ async def send_account_change_recorded( payload: Mapping[str, Any] | BaseModel, idempotency_key: str | None = None, extra_headers: Mapping[str, str] | None = None, + subscriber_id: str | None = None, + through_cursor: str | None = None, ) -> WebhookDeliveryResult: - """Prepare and send a ``account.change_recorded`` notification in one attempt.""" + """Prepare and send a ``account.change_recorded`` notification in one attempt. + + Accepts a change record or notification fields; see + :meth:`prepare_account_change_recorded`. + """ return await self.send_prepared( self.prepare_account_change_recorded( url=url, payload=payload, idempotency_key=idempotency_key, extra_headers=extra_headers, + subscriber_id=subscriber_id, + through_cursor=through_cursor, ) ) diff --git a/tests/test_account_change_feed.py b/tests/test_account_change_feed.py new file mode 100644 index 0000000000..46c0602a2d --- /dev/null +++ b/tests/test_account_change_feed.py @@ -0,0 +1,432 @@ +"""Account change feed records (AdCP 3.3, #1647). + +``changed_paths`` is required on ``adcp`` / ``seller_operator`` records, the +originating request's ``idempotency_key`` reaches only its sender, and a +creative's ``seller_revision`` passes through and mirrors into +``resource_revision``. +""" + +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from adcp._principal_ref import PrincipalRef +from adcp.account_change_feed import ( + ROOT_POINTER, + AccountChangeRecordError, + AttributedAccountChange, + account_change_recorded_payload, + attribute_account_change, + json_pointer, + mirror_creative_revision, + principal_ref_for, + project_account_change, + project_list_account_changes_response, + validate_account_change, +) +from adcp.exceptions import ADCPTaskError +from adcp.server.base import ADCPHandler, ToolContext +from adcp.server.mcp_tools import _requires_changed_paths, create_tool_caller +from adcp.types import ListAccountChangesResponse, ListCreativesResponse +from adcp.types.domains.core.account_change import AccountChange +from adcp.validation.client_hooks import ValidationHookConfig +from adcp.validation.schema_validator import validate_response +from adcp.webhook_sender import WebhookSender + +AT = "2026-10-01T12:00:00Z" +REQUEST_KEY = "req_0123456789abcdef" +URL = "https://buyer.example/webhooks" +STRICT = ValidationHookConfig(requests="strict", responses="strict") + + +def record(**overrides: Any) -> dict[str, Any]: + value: dict[str, Any] = { + "change_id": "chg_1", + "recorded_at": AT, + "resource": {"type": "creative", "account_id": "acct_1", "resource_id": "cr_1"}, + "action": "updated", + "origin": {"kind": "adcp"}, + "idempotency_key": REQUEST_KEY, + "changed_paths": ["/name"], + "repair": {"task": "list_creatives"}, + } + value.update(overrides) + return {key: item for key, item in value.items() if item is not None} + + +def page(changes: list[Any]) -> dict[str, Any]: + return { + "status": "completed", + "changes": changes, + "cursor": "cur_2", + "has_more": False, + "available_since": AT, + "generated_at": AT, + } + + +# -------------------------------------------------------------------------- +# Record rules +# -------------------------------------------------------------------------- + + +def test_json_pointer_escapes_tokens_and_defaults_to_root() -> None: + assert json_pointer() == ROOT_POINTER == "" + assert json_pointer("packages", 0, "budget") == "/packages/0/budget" + assert json_pointer("a/b", "m~n") == "/a~1b/m~0n" + + +@pytest.mark.parametrize("kind", ["adcp", "seller_operator"]) +@pytest.mark.parametrize("paths", [None, []]) +def test_paths_required_for_processed_origins(kind: str, paths: list[str] | None) -> None: + change = record(origin={"kind": kind}, idempotency_key=None, changed_paths=paths) + if paths is not None: + change["changed_paths"] = paths + with pytest.raises(AccountChangeRecordError, match="changed_paths") as exc: + validate_account_change(change, pointer="/changes/2") + assert exc.value.pointer == "/changes/2/changed_paths" + with pytest.raises(AccountChangeRecordError): + attribute_account_change(change, None) + + +@pytest.mark.parametrize("kind", ["seller_system", "connected_platform", "unknown"]) +def test_paths_best_effort_for_observed_origins(kind: str) -> None: + validate_account_change(record(origin={"kind": kind}, idempotency_key=None, changed_paths=None)) + + +def test_root_pointer_satisfies_whole_resource_changes() -> None: + change = record(action="created", changed_paths=[ROOT_POINTER]) + validate_account_change(change) + assert attribute_account_change(change, "p1").changed_paths == [""] + + +def test_idempotency_key_only_on_adcp_records() -> None: + with pytest.raises(AccountChangeRecordError, match="idempotency_key") as exc: + validate_account_change(record(origin={"kind": "seller_operator"})) + assert exc.value.pointer == "/idempotency_key" + + +def test_generated_schema_agrees_with_the_record_rules() -> None: + """The bundled 3.3 schema encodes the same rules the SDK enforces.""" + for change, valid in [ + (record(), True), + (record(changed_paths=None), False), + (record(origin={"kind": "seller_system"}, changed_paths=None, idempotency_key=None), True), + (record(origin={"kind": "seller_system"}, changed_paths=None), False), + ]: + assert validate_response("list_account_changes", page([change])).valid is valid + + +# -------------------------------------------------------------------------- +# Attribution and per-reader projection +# -------------------------------------------------------------------------- + + +def test_principal_ref_for_accepts_context_ref_and_id() -> None: + ctx = ToolContext(caller_identity="p1", tenant_id="t1") + assert principal_ref_for(ctx) == PrincipalRef("t1", "p1") + assert principal_ref_for(ToolContext()) is None + assert principal_ref_for("p1", tenant_id="t1") == PrincipalRef("t1", "p1") + assert principal_ref_for(PrincipalRef(None, "p1")) == PrincipalRef(None, "p1") + assert principal_ref_for(None) is None + + +def test_attribution_is_off_the_wire() -> None: + attributed = attribute_account_change(AccountChange.model_validate(record()), "p1") + assert isinstance(attributed, AttributedAccountChange) + assert attributed.originating_principal == PrincipalRef(None, "p1") + dumped = attributed.model_dump(mode="json", exclude_none=True) + assert "originating_principal" not in json.dumps(dumped) + assert dumped == AccountChange.model_validate(record()).model_dump( + mode="json", exclude_none=True + ) + + +@pytest.mark.parametrize( + ("originator", "reader", "echoed"), + [ + ("p1", PrincipalRef(None, "p1"), True), + ("p1", PrincipalRef(None, "p2"), False), + ("p1", None, False), + (None, PrincipalRef(None, "p1"), False), + (PrincipalRef("t1", "p1"), PrincipalRef("t2", "p1"), False), + (PrincipalRef("t1", "p1"), ToolContext(caller_identity="p1", tenant_id="t1"), True), + (ToolContext(caller_identity="p1"), ToolContext(caller_identity="p1"), True), + ], +) +def test_key_echoed_only_to_the_originator(originator: Any, reader: Any, echoed: bool) -> None: + projected = project_account_change( + attribute_account_change(record(), originator), reader=reader + ) + assert ("idempotency_key" in projected) is echoed + assert projected["changed_paths"] == ["/name"] + + +@pytest.mark.parametrize("change", [record(), AccountChange.model_validate(record())]) +def test_unattributed_records_fail_closed(change: Any) -> None: + projected = project_account_change(change, reader=PrincipalRef(None, "p1")) + assert "idempotency_key" not in projected + assert change is not projected + + +def test_response_projection_keeps_envelope_and_refuses_bad_pages() -> None: + model = ListAccountChangesResponse.model_validate( + {**page([attribute_account_change(record(), "p1")]), "context": {"trace": "x"}} + ) + projected = project_list_account_changes_response(model, reader="p1") + assert projected["context"] == {"trace": "x"} + assert projected["changes"][0]["idempotency_key"] == REQUEST_KEY + + bad = page([record(change_id="chg_0"), record(changed_paths=None)]) + with pytest.raises(AccountChangeRecordError) as exc: + project_list_account_changes_response(bad, reader="p1") + assert exc.value.pointer == "/changes/1/changed_paths" + loose = project_list_account_changes_response(bad, reader="p1", require_changed_paths=False) + assert all("idempotency_key" not in change for change in loose["changes"]) + + error_envelope = {"adcp_error": {"code": "ACCOUNT_NOT_FOUND"}} + assert project_list_account_changes_response(error_envelope, reader="p1") is error_envelope + + +# -------------------------------------------------------------------------- +# Server dispatch +# -------------------------------------------------------------------------- + + +class FeedSeller(ADCPHandler): + def __init__(self, changes: list[Any], *, as_model: bool = True) -> None: + self.changes = changes + self.as_model = as_model + + async def list_account_changes(self, params: Any, context: Any = None) -> Any: + if self.as_model: + return ListAccountChangesResponse.model_validate(page(self.changes)) + return page(self.changes) + + +REQUEST = {"account": {"account_id": "acct_1"}} + + +def make_caller(handler: ADCPHandler, tool: str, **kwargs: Any) -> Any: + # Without a version envelope, serve the current release (as A2A does). + return create_tool_caller(handler, tool, default_unnegotiated_adcp_version=None, **kwargs) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("as_model", [True, False]) +async def test_dispatch_echoes_key_only_to_the_originating_caller(as_model: bool) -> None: + mine = attribute_account_change(record(change_id="chg_mine"), "buyer_a") + theirs = attribute_account_change(record(change_id="chg_theirs"), "buyer_b") + plain = record(change_id="chg_plain") + # A key on a non-adcp record is dropped, not refused. + operator = record(change_id="chg_op", origin={"kind": "seller_operator"}) + caller = make_caller( + FeedSeller([mine, theirs, plain, operator], as_model=as_model), + "list_account_changes", + validation=STRICT, + ) + + def keys(response: dict[str, Any]) -> dict[str, str | None]: + return {c["change_id"]: c.get("idempotency_key") for c in response["changes"]} + + as_a = await caller(dict(REQUEST), ToolContext(caller_identity="buyer_a")) + assert keys(as_a) == { + "chg_mine": REQUEST_KEY, + "chg_theirs": None, + "chg_plain": None, + "chg_op": None, + } + as_b = await caller(dict(REQUEST), ToolContext(caller_identity="buyer_b")) + assert keys(as_b)["chg_theirs"] == REQUEST_KEY + assert keys(as_b)["chg_mine"] is None + anonymous = await caller(dict(REQUEST), ToolContext()) + assert set(keys(anonymous).values()) == {None} + other_tenant = await caller( + dict(REQUEST), ToolContext(caller_identity="buyer_a", tenant_id="t9") + ) + assert set(keys(other_tenant).values()) == {None} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("validation", [STRICT, None]) +async def test_dispatch_refuses_record_without_changed_paths( + validation: ValidationHookConfig | None, +) -> None: + bad = record(origin={"kind": "seller_operator"}, idempotency_key=None, changed_paths=None) + caller = make_caller( + FeedSeller([record(), bad], as_model=False), "list_account_changes", validation=validation + ) + with pytest.raises(ADCPTaskError) as exc: + await caller(dict(REQUEST), ToolContext(caller_identity="buyer_a")) + error = exc.value.errors[0] + assert error.code == "VALIDATION_ERROR" + assert error.field == "/changes/1/changed_paths" + assert "changed_paths" in error.message + + +def test_changed_paths_enforced_from_3_3() -> None: + assert _requires_changed_paths("3.3-beta.3") + assert _requires_changed_paths("3.3") + assert _requires_changed_paths(None) + assert not _requires_changed_paths("3.2") + + +@pytest.mark.asyncio +async def test_dispatch_on_a_3_2_handler_keeps_paths_optional_but_strips_keys() -> None: + class Seller32(FeedSeller): + _adcp_version = "3.2" + + plain = record(changed_paths=None) + caller = make_caller( + Seller32([plain], as_model=False), "list_account_changes", supported_versions=["3.2"] + ) + response = await caller({**REQUEST, "adcp_version": "3.2"}, ToolContext(caller_identity="x")) + assert "changed_paths" not in response["changes"][0] + assert "idempotency_key" not in response["changes"][0] + + +# -------------------------------------------------------------------------- +# account.change_recorded notifications +# -------------------------------------------------------------------------- + + +def test_notification_payload_never_carries_the_request_key() -> None: + change = attribute_account_change( + record(resource={**record()["resource"], "parent_ids": {"media_buy_id": "mb_1"}}), "p1" + ) + payload = account_change_recorded_payload(change, subscriber_id="sub_1", through_cursor="c9") + assert payload == { + "notification_type": "account.change_recorded", + "notification_id": "chg_1", + "subscriber_id": "sub_1", + "account_id": "acct_1", + "change_id": "chg_1", + "recorded_at": AT, + "resource": { + "type": "creative", + "resource_id": "cr_1", + "parent_ids": {"media_buy_id": "mb_1"}, + }, + "action": "updated", + "through_cursor": "c9", + } + + +@pytest.mark.parametrize("change", [record(), AccountChange.model_validate(record())]) +def test_prepare_from_a_record_mints_a_fresh_delivery_key(change: Any) -> None: + prepared = WebhookSender.prepare_account_change_recorded( + url=URL, payload=change, subscriber_id="sub_1" + ) + body = json.loads(prepared.body) + assert REQUEST_KEY not in prepared.body.decode() + assert body["idempotency_key"] == prepared.idempotency_key != REQUEST_KEY + assert body["notification_id"] == body["change_id"] == "chg_1" + assert "changed_paths" not in body and "origin" not in body + + explicit = WebhookSender.prepare_account_change_recorded( + url=URL, payload=change, subscriber_id="sub_1", idempotency_key="delivery_0123456789" + ) + assert explicit.idempotency_key == "delivery_0123456789" + + +def test_prepare_refuses_non_conformant_records_and_key_reuse() -> None: + with pytest.raises(AccountChangeRecordError, match="changed_paths"): + WebhookSender.prepare_account_change_recorded( + url=URL, payload=record(changed_paths=[]), subscriber_id="sub_1" + ) + with pytest.raises(ValueError, match="must not reuse"): + WebhookSender.prepare_account_change_recorded( + url=URL, payload=record(), subscriber_id="sub_1", idempotency_key=REQUEST_KEY + ) + with pytest.raises(ValueError, match="subscriber_id"): + WebhookSender.prepare_account_change_recorded(url=URL, payload=record()) + + +def test_prepare_from_notification_fields_is_unchanged() -> None: + fields = account_change_recorded_payload(record(), subscriber_id="sub_1") + prepared = WebhookSender.prepare_account_change_recorded( + url=URL, payload=fields, idempotency_key="delivery_0123456789" + ) + assert json.loads(prepared.body)["idempotency_key"] == "delivery_0123456789" + with pytest.raises(ValueError, match="only when payload is a change record"): + WebhookSender.prepare_account_change_recorded(url=URL, payload=fields, subscriber_id="s") + + +# -------------------------------------------------------------------------- +# Creative seller_revision +# -------------------------------------------------------------------------- + + +def creatives_page(seller_revision: Any) -> dict[str, Any]: + return { + "status": "completed", + "query_summary": {"total_matching": 1, "returned": 1}, + "pagination": {"has_more": False}, + "creatives": [ + { + "creative_id": "cr_1", + "seller_revision": seller_revision, + "name": "Spring", + "format_kind": "image", + "status": "approved", + "created_date": AT, + "updated_date": AT, + } + ], + } + + +@pytest.mark.parametrize("revision", [7, "rev-7"]) +def test_seller_revision_round_trips_the_model(revision: Any) -> None: + model = ListCreativesResponse.model_validate(creatives_page(revision)) + dumped = model.model_dump(mode="json", exclude_none=True) + assert dumped["creatives"][0]["seller_revision"] == revision + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("version", "kept"), [(None, True), ("3.2", False)]) +async def test_list_creatives_serves_seller_revision_to_3_3_only( + version: str | None, kept: bool +) -> None: + class CreativeSeller(ADCPHandler): + async def list_creatives(self, params: Any, context: Any = None) -> Any: + return ListCreativesResponse.model_validate(creatives_page(7)) + + caller = make_caller(CreativeSeller(), "list_creatives", validation=STRICT) + wire: dict[str, Any] = {} if version is None else {"adcp_version": version} + response = await caller(wire, ToolContext(caller_identity="buyer_a")) + assert response["creatives"][0].get("seller_revision") == (7 if kept else None) + + +@pytest.mark.parametrize("revision", [7, "rev-7"]) +def test_mirror_creative_revision_keeps_json_type(revision: Any) -> None: + creative = ListCreativesResponse.model_validate(creatives_page(revision)).creatives[0] + attributed = attribute_account_change(record(), "p1") + mirrored = mirror_creative_revision(attributed, creative) + assert isinstance(mirrored, AttributedAccountChange) + assert mirrored.originating_principal == PrincipalRef(None, "p1") + assert mirrored.model_dump(mode="json")["resource_revision"] == revision + assert attributed.resource_revision is None + + as_dict = mirror_creative_revision(record(), {"seller_revision": revision}) + assert as_dict["resource_revision"] == revision + assert type(as_dict["resource_revision"]) is type(revision) + + +def test_mirror_creative_revision_clears_and_validates() -> None: + assert "resource_revision" not in mirror_creative_revision(record(resource_revision=3), {}) + cleared = mirror_creative_revision( + AccountChange.model_validate(record(resource_revision=3)), {} + ) + assert "resource_revision" not in cleared.model_dump(mode="json", exclude_none=True) + with pytest.raises(AccountChangeRecordError, match="not creative"): + mirror_creative_revision( + record(resource={"type": "media_buy", "account_id": "a", "resource_id": "m"}), + {"seller_revision": 1}, + ) + for bad in (0, True, "", "x" * 256, 1.5): + with pytest.raises(AccountChangeRecordError): + mirror_creative_revision(record(), {"seller_revision": bad})