From bc1c247eb0b6f2bc4d36edcb8e9a3714f1ef62b1 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 18:12:35 +0000 Subject: [PATCH] feat(reporting): add the buyer write facade and guard raw sync_principal sync_principal treats reporting_subscriptions and reporting_destinations as complete desired state, so a raw write that adds one feed deactivates every other subscription, and omitting a destination revokes it for every account. Python had no guard (DX plan section 4, item 1). client.reporting (ReportingClient) is patch-shaped: - subscribe() replaces one subscription and keeps every other entry as the seller stores it. It auto-versions: the version is kept when only active or replace_account_configs change and bumped when the selector, scope, destination, reconciliation_mode, exclude_account_ids or operator_domains change; a supplied version must exceed the current one when the content changed. register_destination upserts a destination in the same write. An unchanged subscription writes nothing. - unsubscribe() is the only deactivation path; resubscribing reactivates at the same version. - replace_subscriptions() is the expert complete-set write: omitting an active subscription needs deactivate_omitted=True, and [] needs deactivate_all=True. - list_subscriptions() and preview() (a seller dry run) round it out. Pre-dispatch checks raise ReportingSubscriptionError with dispatched=False: an explicit scope, a registered active destination, {account_id} in the path template of a multi-account scope, a subscription_id that clashes with a delivery_config_id on list_accounts (SUBSCRIPTION_ID_CLASH), an all-accounts scope across operators without operator_domains (SUBSCRIPTION_OPERATOR_SCOPE_REQUIRED), and a selector no advertised offering matches (SUBSCRIPTION_SELECTOR_UNMATCHED, naming the nearest offerings). The roster and selector checks run when the client can read list_accounts and capabilities; a partial roster is never used. Each write carries expected_configuration_version and a fresh idempotency key per attempt. A transport failure resends the same key and body once; CONFLICT re-reads, re-merges and retries, up to 3 attempts, then raises SUBSCRIPTION_CONFLICT_EXHAUSTED. The readback must carry every submitted subscription at its version, state and content, or the write raises SUBSCRIPTION_READBACK_MISMATCH with the readback attached. ADCPClient.sync_principal now reads the principal first when a write carries either section, and refuses before dispatch a write that would deactivate an active subscription (SUBSCRIPTION_WOULD_DEACTIVATE) or revoke a destination (DESTINATION_WOULD_REVOKE) by omission, unless allow_deactivation / allow_revocation is passed. An explicit active: false is always allowed. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/client.py | 45 +- src/adcp/reporting/subscriptions/__init__.py | 10 + src/adcp/reporting/subscriptions/consumer.py | 5 +- src/adcp/reporting/subscriptions/facade.py | 818 +++++++++++++++++++ tests/test_reporting_subscription_facade.py | 346 ++++++++ 5 files changed, 1222 insertions(+), 2 deletions(-) create mode 100644 src/adcp/reporting/subscriptions/facade.py create mode 100644 tests/test_reporting_subscription_facade.py diff --git a/src/adcp/client.py b/src/adcp/client.py index cdcdabc2e..0eb77261a 100644 --- a/src/adcp/client.py +++ b/src/adcp/client.py @@ -26,6 +26,8 @@ import httpx from mcp import ClientSession + from adcp.reporting.subscriptions.facade import ReportingClient + from adcp._null_clear import preserve_explicit_nulls from adcp._response_limit import ( DEFAULT_MAX_RESPONSE_BYTES, @@ -962,6 +964,7 @@ def __init__( self._raise_account_errors_explicit = raise_account_errors is not None self.raise_account_errors = raise_account_errors is True self.accounts = AccountRegistry(self, account_storage) + self._reporting: Any | None = None self.webhook_url_template = webhook_url_template self.webhook_secret = webhook_secret self.allow_unauthenticated_webhooks = allow_unauthenticated_webhooks @@ -1322,6 +1325,18 @@ def get_server_version(self) -> str | None: """ return self._server_version + @property + def reporting(self) -> ReportingClient: + """Patch-shaped reporting-subscription writes: ``subscribe``, ``unsubscribe``, ... + + See :class:`adcp.reporting.subscriptions.facade.ReportingClient`. + """ + if self._reporting is None: + from adcp.reporting.subscriptions.facade import ReportingClient + + self._reporting = ReportingClient(self) + return cast("ReportingClient", self._reporting) + @property def context_id(self) -> str | None: """Current A2A conversation context_id. @@ -4434,9 +4449,37 @@ async def sync_principal( self, request: SyncPrincipalRequest, *, + allow_deactivation: bool = False, + allow_revocation: bool = False, options: TaskOptions | None = None, ) -> TaskResult[SyncPrincipalResponse]: - """Declaratively replace selected authenticated-principal configuration.""" + """Declaratively replace selected authenticated-principal configuration. + + Each section is complete desired state: a subscription missing from + ``reporting_subscriptions`` is deactivated, and a destination missing + from ``reporting_destinations`` is revoked for every account. To add + or change one subscription, use ``client.reporting.subscribe()``. + + A write carrying either section first reads the principal and is + refused before dispatch (``ReportingSubscriptionError``, code + ``SUBSCRIPTION_WOULD_DEACTIVATE`` or ``DESTINATION_WOULD_REVOKE``) if it + would deactivate an active subscription or revoke a destination by + omission. Pass ``allow_deactivation=True`` / ``allow_revocation=True`` + to proceed. Sending ``active: false`` is always allowed. + """ + from adcp.reporting.subscriptions.facade import guard_sync_principal + + await guard_sync_principal( + self, + request, + allow_deactivation=allow_deactivation, + allow_revocation=allow_revocation, + ) + return await self._sync_principal_unguarded(request) + + async def _sync_principal_unguarded( + self, request: SyncPrincipalRequest + ) -> TaskResult[SyncPrincipalResponse]: return cast( TaskResult[SyncPrincipalResponse], await self._execute_typed_task("sync_principal", request, SyncPrincipalResponse), diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index 6e0488b1b..b20b4fdbb 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -79,6 +79,12 @@ SubscriptionPlanResult, subscription_obligation_id, ) +from adcp.reporting.subscriptions.facade import ( + PreviewResult, + ReportingClient, + SubscribeResult, + guard_sync_principal, +) from adcp.reporting.subscriptions.registry import ( ReportingSubscriptionValidationError, apply_reporting_subscriptions, @@ -103,6 +109,10 @@ ) __all__ = [ + "PreviewResult", + "ReportingClient", + "SubscribeResult", + "guard_sync_principal", "FrozenOwedSet", "SubscriptionObligationPlanner", "SubscriptionPlanResult", diff --git a/src/adcp/reporting/subscriptions/consumer.py b/src/adcp/reporting/subscriptions/consumer.py index 8adc40edc..fab5df750 100644 --- a/src/adcp/reporting/subscriptions/consumer.py +++ b/src/adcp/reporting/subscriptions/consumer.py @@ -25,7 +25,7 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from datetime import datetime, timedelta -from typing import Any, Protocol +from typing import Any, Literal, Protocol from uuid import uuid4 from adcp.exceptions import ADCPError, RecoveryLiteral, _normalize_recovery @@ -95,6 +95,7 @@ def __init__( *, code: str | None = None, recovery: RecoveryLiteral | None = None, + dispatched: Literal[False] | None = None, would_deactivate: Sequence[str] = (), readback: ReportingSubscriptionReadback | None = None, ) -> None: @@ -107,6 +108,8 @@ def __init__( self.dispatched = False else: self.recovery = recovery or _code_recovery(code) + if dispatched is False: + self.dispatched = False self.would_deactivate = tuple(would_deactivate) self.readback = readback diff --git a/src/adcp/reporting/subscriptions/facade.py b/src/adcp/reporting/subscriptions/facade.py new file mode 100644 index 000000000..a019f48a5 --- /dev/null +++ b/src/adcp/reporting/subscriptions/facade.py @@ -0,0 +1,818 @@ +"""The buyer's reporting-subscription write facade: ``client.reporting``. + +``sync_principal`` treats ``reporting_subscriptions`` and +``reporting_destinations`` as complete desired state, so a raw write that +"adds one feed" deactivates every other subscription, and one that omits a +destination revokes it for every account. This facade is patch-shaped: + +- :meth:`ReportingClient.subscribe` replaces one subscription and keeps every + other entry as the seller stores it. It chooses ``subscription_version``, + checks the request before sending it, writes under + ``expected_configuration_version`` and verifies the readback. +- :meth:`ReportingClient.unsubscribe` is the only deactivation path. +- :meth:`ReportingClient.replace_subscriptions` is the expert complete-set + write; deactivating what it omits needs ``deactivate_omitted=True``. + +:func:`guard_sync_principal` is the guard ``ADCPClient.sync_principal`` +applies to raw writes that carry either section. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable, Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any, Literal +from uuid import uuid4 + +from adcp.reporting.subscriptions.consumer import ( + PrincipalClient, + ReportingSubscriptionError, + ReportingSubscriptionReadback, + _dump, +) +from adcp.reporting.subscriptions.registry import _semantic +from adcp.types import GetPrincipalRequest, SyncPrincipalRequest + +SubscribeAction = Literal["created", "updated", "unchanged", "reactivated", "deactivated"] + +#: Conflict retries with a fresh key, and the bound on the roster read. +_DEFAULT_ATTEMPTS = 3 +_ROSTER_PAGES = 10 + + +@dataclass(frozen=True) +class SubscribeResult: + """The outcome of a facade write for one subscription.""" + + action: SubscribeAction + #: The seller's readback for this subscription. + state: Any + #: The version the SDK chose (or kept). + subscription_version: int + #: The readback of the destination the subscription names, if any. + destination: Any | None + readback: ReportingSubscriptionReadback + + +@dataclass(frozen=True) +class PreviewResult: + """What :meth:`ReportingClient.subscribe` would do, validated by the seller (dry run).""" + + would_be: SubscribeAction + subscription_version: int + #: Offerings the selector resolves to among the seller's advertised ones. + offering_ids: tuple[str, ...] = () + + +@dataclass(frozen=True) +class _Principal: + """The caller's current principal: what a write replaces.""" + + kind: str + configuration_version: str | None + principal_kind: str | None + subscriptions: tuple[Any, ...] = () + destinations: tuple[Any, ...] = () + + def subscription(self, subscription_id: str) -> Any | None: + return next( + (s for s in self.subscriptions if _value(s, "subscription_id") == subscription_id), + None, + ) + + def submitted_subscriptions(self) -> list[dict[str, Any]]: + """Every subscription as submitted (its current version's configuration).""" + return [_dump(_value(item, "configuration")) for item in self.subscriptions] + + def submitted_destinations(self) -> list[dict[str, Any]]: + return [_dump(_value(item, "configuration")) for item in self.destinations] + + +def _value(item: Any, name: str, default: Any = None) -> Any: + if isinstance(item, Mapping): + return item.get(name, default) + return getattr(item, name, default) + + +def _text(value: Any) -> str: + return str(getattr(value, "value", value) if value is not None else "") + + +def _active_subscription(state: Any) -> bool: + return _text(_value(state, "state")) != "inactive" + + +def _active_destination(configuration: Mapping[str, Any]) -> bool: + return configuration.get("active", True) is not False + + +def _refused(message: str, code: str, **kwargs: Any) -> ReportingSubscriptionError: + return ReportingSubscriptionError(message, code=code, **kwargs) + + +def _invalid(message: str) -> ReportingSubscriptionError: + """A request the SDK refuses before sending; the fix ends the message.""" + return ReportingSubscriptionError(message, code="INVALID_REQUEST", dispatched=False) + + +async def read_principal(client: PrincipalClient) -> _Principal: + result = await client.get_principal(GetPrincipalRequest.model_validate({})) + response = getattr(result, "data", None) + if not getattr(result, "success", False) or response is None: + raise ReportingSubscriptionError( + f"get_principal failed: {getattr(result, 'error', None) or 'no response'}. " + "Retry once the seller is reachable." + ) + outcome = response.result + kind = _text(getattr(outcome, "kind", "")) + if kind == "failed": + errors = list(getattr(outcome, "errors", None) or []) + first = errors[0] if errors else None + raise ReportingSubscriptionError( + f"get_principal was rejected: {getattr(first, 'message', 'unknown error')}", + code=None if first is None else (_text(getattr(first, "code", "")) or None), + ) + configuration = getattr(outcome, "configuration", None) + return _Principal( + kind=kind, + configuration_version=getattr(outcome, "configuration_version", None), + principal_kind=_text(getattr(outcome, "principal_kind", None)) or None, + subscriptions=tuple(getattr(configuration, "reporting_subscriptions", None) or ()), + destinations=tuple(getattr(configuration, "reporting_destinations", None) or ()), + ) + + +def _deactivated_by(principal: _Principal, subscriptions: Sequence[Mapping[str, Any]]) -> list[str]: + """Active subscriptions a complete-set write omits (explicit ``active: false`` is allowed).""" + sent = {item.get("subscription_id") for item in subscriptions} + return [ + str(_value(state, "subscription_id")) + for state in principal.subscriptions + if _active_subscription(state) and _value(state, "subscription_id") not in sent + ] + + +def _revoked_by(principal: _Principal, destinations: Sequence[Mapping[str, Any]]) -> list[str]: + """Active destinations a complete-set write omits.""" + sent = {item.get("destination_id") for item in destinations} + return [ + str(item.get("destination_id")) + for item in principal.submitted_destinations() + if _active_destination(item) and item.get("destination_id") not in sent + ] + + +def check_complete_set( + principal: _Principal, + configuration: Mapping[str, Any], + *, + allow_deactivation: bool, + allow_revocation: bool, +) -> None: + """Refuse, before sending, a complete-set write that deactivates or revokes by omission.""" + if "reporting_subscriptions" in configuration and not allow_deactivation: + omitted = _deactivated_by(principal, configuration["reporting_subscriptions"] or []) + if omitted: + raise _refused( + "This write omits active subscriptions " + f"{', '.join(repr(i) for i in omitted)}, which deactivates them. " + "Use client.reporting.subscribe() or unsubscribe(), or pass " + "allow_deactivation=True.", + "SUBSCRIPTION_WOULD_DEACTIVATE", + would_deactivate=omitted, + ) + if "reporting_destinations" in configuration and not allow_revocation: + revoked = _revoked_by(principal, configuration["reporting_destinations"] or []) + if revoked: + raise _refused( + f"This write omits destinations {', '.join(repr(i) for i in revoked)}. " + "Omitting a destination revokes it for every account. Pass " + "allow_revocation=True.", + "DESTINATION_WOULD_REVOKE", + ) + + +async def guard_sync_principal( + client: PrincipalClient, + request: Any, + *, + allow_deactivation: bool = False, + allow_revocation: bool = False, +) -> None: + """The raw ``sync_principal`` guard: one read, only for writes that touch either section.""" + configuration = _dump(_value(request, "configuration") or {}) + touched = {"reporting_subscriptions", "reporting_destinations"} & set(configuration) + if not touched or _value(request, "dry_run") is True: + return + if (allow_deactivation or "reporting_subscriptions" not in touched) and ( + allow_revocation or "reporting_destinations" not in touched + ): + return + check_complete_set( + await read_principal(client), + configuration, + allow_deactivation=allow_deactivation, + allow_revocation=allow_revocation, + ) + + +def differing_path(sent: Any, stored: Any, path: str = "") -> str | None: + """The first JSON Pointer at which ``stored`` lacks or changes a value the buyer sent. + + Fields the seller adds (defaults it fills) are ignored; arrays must match + element for element. + """ + if isinstance(sent, list): + if not isinstance(stored, list) or len(stored) != len(sent): + return path or "/" + for index, item in enumerate(sent): + found = differing_path(item, stored[index], f"{path}/{index}") + if found is not None: + return found + return None + if isinstance(sent, dict): + if not isinstance(stored, dict): + return path or "/" + for key, value in sent.items(): + if value is None: + continue + found = differing_path(value, stored.get(key), f"{path}/{key}") + if found is not None: + return found + return None + return None if sent == stored else (path or "/") + + +class ReportingClient: + """``client.reporting``: patch-shaped reporting-subscription writes. + + ``client`` needs ``get_principal`` and ``sync_principal``. With + ``list_accounts`` and ``fetch_capabilities`` (an ``ADCPClient``), the + subscription-id clash, operator and selector checks also run before + anything is sent. + """ + + def __init__( + self, + client: Any, + *, + max_attempts: int = _DEFAULT_ATTEMPTS, + idempotency_key: Callable[[], str] | None = None, + ) -> None: + if type(max_attempts) is not int or not 1 <= max_attempts <= 10: + raise ValueError("max_attempts must be an integer in 1..10") + self._client = client + self._attempts = max_attempts + self._key = idempotency_key or (lambda: f"rsub-{uuid4()}") + + # -- public API --------------------------------------------------------------- + + async def list_subscriptions(self) -> ReportingSubscriptionReadback: + """The caller's subscriptions as the seller reports them (``get_principal``).""" + principal = await read_principal(self._client) + return ReportingSubscriptionReadback( + configuration_version=principal.configuration_version, + subscriptions=principal.subscriptions, + ) + + async def subscribe( + self, + subscription: Mapping[str, Any] | Any, + *, + register_destination: Mapping[str, Any] | Any | None = None, + acknowledge_cross_operator: bool = False, + ) -> SubscribeResult: + """Create or update one subscription, keeping every other entry unchanged. + + ``subscription_version`` is optional: the SDK keeps it when only + ``active`` or ``replace_account_configs`` change, and bumps it when + the selector, scope, destination, ``reconciliation_mode``, + ``exclude_account_ids`` or ``operator_domains`` change. A supplied + version must be greater than the current one when the content + changed. ``register_destination`` upserts a destination in the same + atomic write. + """ + wanted = _dump(subscription) + destination = None if register_destination is None else _dump(register_destination) + + async def build(principal: _Principal) -> _Plan: + return await self._plan_subscribe( + principal, + wanted, + destination, + acknowledge_cross_operator=acknowledge_cross_operator, + ) + + return await self._write(build) + + async def unsubscribe(self, subscription_id: str) -> SubscribeResult: + """Deactivate one subscription (``active: false``); its obligations are retained.""" + + async def build(principal: _Principal) -> _Plan: + state = principal.subscription(subscription_id) + if state is None: + raise _invalid( + f"No subscription {subscription_id!r} exists for this principal. " + "Check list_subscriptions() for the ids." + ) + current = _dump(_value(state, "configuration")) + if not _active_subscription(state): + return _Plan.unchanged(principal, current) + return self._plan_replace_one(principal, {**current, "active": False}, "deactivated") + + return await self._write(build) + + async def preview( + self, + subscription: Mapping[str, Any] | Any, + *, + register_destination: Mapping[str, Any] | Any | None = None, + acknowledge_cross_operator: bool = False, + ) -> PreviewResult: + """Validate what :meth:`subscribe` would write, as a seller dry run.""" + principal = await read_principal(self._client) + plan = await self._plan_subscribe( + principal, + _dump(subscription), + None if register_destination is None else _dump(register_destination), + acknowledge_cross_operator=acknowledge_cross_operator, + ) + if plan.configuration is not None: + await self._send(principal, plan.configuration, key=self._key(), dry_run=True) + return PreviewResult(plan.action, plan.version, plan.offering_ids) + + async def replace_subscriptions( + self, + subscriptions: Sequence[Mapping[str, Any] | Any], + *, + deactivate_omitted: bool = False, + deactivate_all: bool = False, + reporting_destinations: Sequence[Mapping[str, Any] | Any] | None = None, + allow_revocation: bool = False, + ) -> ReportingSubscriptionReadback: + """Replace the complete subscription set (expert). + + Omitting an active subscription deactivates it, which needs + ``deactivate_omitted=True``; an empty list needs ``deactivate_all=True``. + Versions are taken as given. + """ + sent = [_dump(item) for item in subscriptions] + if not sent and not deactivate_all: + raise _invalid( + "An empty subscription list deactivates every subscription. " + "Pass deactivate_all=True." + ) + destinations = ( + None if reporting_destinations is None else [_dump(i) for i in reporting_destinations] + ) + + async def build(principal: _Principal) -> _Plan: + configuration: dict[str, Any] = {"reporting_subscriptions": sent} + if destinations is not None: + configuration["reporting_destinations"] = destinations + check_complete_set( + principal, + configuration, + allow_deactivation=deactivate_omitted or deactivate_all, + allow_revocation=allow_revocation, + ) + return _Plan(configuration, "updated", 0, None, sent) + + plan_result = await self._write(build, single=False) + return plan_result.readback + + # -- planning ------------------------------------------------------------------- + + async def _plan_subscribe( + self, + principal: _Principal, + wanted: dict[str, Any], + destination: dict[str, Any] | None, + *, + acknowledge_cross_operator: bool, + ) -> _Plan: + subscription_id = wanted.get("subscription_id") + if not isinstance(subscription_id, str) or not subscription_id: + raise _invalid("subscribe() needs a subscription_id. Name the subscription.") + wanted.setdefault("active", True) + state = principal.subscription(subscription_id) + current = None if state is None else _dump(_value(state, "configuration")) + version, action = self._choose_version(wanted, current, state) + wanted["subscription_version"] = version + destinations = principal.submitted_destinations() + destinations_changed = False + if destination is not None: + merged = [ + d + for d in destinations + if d.get("destination_id") != destination.get("destination_id") + ] + [destination] + if merged != destinations and destination not in destinations: + destinations, destinations_changed = merged, True + self._check_scope(wanted) + named = self._check_destination(wanted, destinations) + offering_ids = await self._check_seller( + wanted, named, acknowledge_cross_operator=acknowledge_cross_operator + ) + if action == "unchanged" and not destinations_changed: + plan = _Plan.unchanged(principal, wanted) + plan.offering_ids = offering_ids + return plan + plan = self._plan_replace_one(principal, wanted, action) + if destinations_changed: + assert plan.configuration is not None + plan.configuration["reporting_destinations"] = destinations + plan.offering_ids = offering_ids + return plan + + @staticmethod + def _choose_version( + wanted: dict[str, Any], current: dict[str, Any] | None, state: Any | None + ) -> tuple[int, SubscribeAction]: + supplied = wanted.get("subscription_version") + if supplied is not None and (type(supplied) is not int or supplied < 1): + raise _invalid( + "subscription_version must be a positive integer. Omit it to auto-version." + ) + if current is None: + return supplied or 1, "created" + version = int( + _value(state, "subscription_version") or current.get("subscription_version") or 1 + ) + reactivating = not _active_subscription(state) and wanted.get("active") is not False + if _semantic({**current, "subscription_id": wanted["subscription_id"]}) != _semantic( + wanted + ): + if supplied is not None and supplied <= version: + raise _invalid( + f"Subscription {wanted['subscription_id']!r} changes its content, so its " + f"version must exceed {version}. Omit subscription_version to auto-version." + ) + return supplied or version + 1, "reactivated" if reactivating else "updated" + if supplied is not None and supplied != version: + raise _invalid( + f"Subscription {wanted['subscription_id']!r} is unchanged at version {version}. " + "Omit subscription_version, or change the content." + ) + if reactivating: + return version, "reactivated" + if (wanted.get("active") is False) != (not _active_subscription(state)) or wanted.get( + "replace_account_configs" + ) != current.get("replace_account_configs"): + return version, "updated" + return version, "unchanged" + + @staticmethod + def _plan_replace_one( + principal: _Principal, wanted: dict[str, Any], action: SubscribeAction + ) -> _Plan: + subscriptions = [ + wanted if item.get("subscription_id") == wanted["subscription_id"] else item + for item in principal.submitted_subscriptions() + ] + if not any(item is wanted for item in subscriptions): + subscriptions.append(wanted) + return _Plan( + {"reporting_subscriptions": subscriptions}, + action, + int(wanted["subscription_version"]), + wanted["subscription_id"], + [wanted], + ) + + @staticmethod + def _check_scope(wanted: Mapping[str, Any]) -> None: + scope = wanted.get("scope") or {} + if scope.get("all_authorized_accounts") is not True and not scope.get("account_ids"): + raise _invalid( + "The subscription scope must be explicit. Pass " + '{"all_authorized_accounts": True} or {"account_ids": [...]}.' + ) + + @staticmethod + def _check_destination( + wanted: Mapping[str, Any], destinations: Sequence[Mapping[str, Any]] + ) -> dict[str, Any] | None: + target = wanted.get("destination") + if not target: + return None + destination_id = target.get("destination_id") + named = next((d for d in destinations if d.get("destination_id") == destination_id), None) + if named is None or not _active_destination(named): + raise _invalid( + f"Destination {destination_id!r} is not registered and active. Pass it as " + "register_destination=..." + ) + scope = wanted.get("scope") or {} + several = ( + scope.get("all_authorized_accounts") is True or len(scope.get("account_ids") or ()) > 1 + ) + template = target.get("path_template") + if several and isinstance(template, str) and "{account_id}" not in template: + raise _invalid( + "This subscription can cover several accounts, so its path_template must " + "contain {account_id}. Add it to the template." + ) + return dict(named) + + async def _check_seller( + self, + wanted: Mapping[str, Any], + destination: Mapping[str, Any] | None, + *, + acknowledge_cross_operator: bool, + ) -> tuple[str, ...]: + """Checks that need the seller's roster or capabilities; skipped when unavailable.""" + accounts = await self._roster() + if accounts is not None: + subscription_id = wanted["subscription_id"] + for account in accounts: + configs = _value(account, "reporting_delivery_configs") or () + if any( + _value(_value(c, "configuration"), "delivery_config_id", None) + == subscription_id + or _value(c, "delivery_config_id") == subscription_id + for c in configs + ): + raise _refused( + f"subscription_id {subscription_id!r} is already a delivery_config_id " + f"on account {_value(account, 'account_id')!r}. Choose another " + "subscription_id.", + "SUBSCRIPTION_ID_CLASH", + ) + scope = wanted.get("scope") or {} + if ( + scope.get("all_authorized_accounts") is True + and not wanted.get("operator_domains") + and not acknowledge_cross_operator + ): + operators = { + _text(_value(_value(account, "operator"), "domain", _value(account, "operator"))) + for account in accounts or () + if _value(account, "operator") is not None + } + if len(operators) > 1 or (destination or {}).get("operator_id"): + raise _refused( + "An all-accounts subscription follows every grant, including accounts " + "another operator grants later, into one destination. Add " + "operator_domains, or pass acknowledge_cross_operator=True.", + "SUBSCRIPTION_OPERATOR_SCOPE_REQUIRED", + ) + return await self._resolve_selector(wanted, destination) + + async def _roster(self) -> list[Any] | None: + list_accounts = getattr(self._client, "list_accounts", None) + if list_accounts is None: + return None + from adcp.types import ListAccountsRequest + + accounts: list[Any] = [] + cursor: str | None = None + try: + for _ in range(_ROSTER_PAGES): + payload: dict[str, Any] = {} + if cursor: + payload["pagination"] = {"cursor": cursor} + result = await list_accounts(ListAccountsRequest.model_validate(payload)) + data = getattr(result, "data", None) + if not getattr(result, "success", False) or data is None: + return None + accounts.extend(getattr(data, "accounts", None) or ()) + pagination = getattr(data, "pagination", None) + cursor = getattr(pagination, "cursor", None) + if not getattr(pagination, "has_more", False) or not cursor: + return accounts + except Exception: # noqa: BLE001 - the roster checks are best effort + return None + return None # a partial roster is never used + + async def _resolve_selector( + self, wanted: Mapping[str, Any], destination: Mapping[str, Any] | None + ) -> tuple[str, ...]: + fetch = getattr(self._client, "fetch_capabilities", None) + if fetch is None: + return () + try: + capabilities = await fetch() + dumped = capabilities.model_dump(mode="json", exclude_none=True) + except Exception: # noqa: BLE001 - the selector check is best effort + return () + raw = (dumped.get("media_buy") or {}).get("reporting_delivery") or {} + from adcp.reporting.subscriptions.resolution import ( + SubscriptionOffering, + resolve_subscription_offerings, + ) + + offerings = [] + for item in raw.get("offerings") or (): + try: + offerings.append(SubscriptionOffering.from_offering(item)) + except (TypeError, ValueError): + continue + if not offerings: + return () + try: + resolution = resolve_subscription_offerings( + wanted, + offerings, + product_offering_ids=[o.offering_id for o in offerings], + destination=destination, + ) + except (TypeError, ValueError): + return () + if resolution.offering_ids: + return resolution.offering_ids + selector = wanted.get("selector") or {} + nearest = ", ".join( + f"{o.offering_id} ({_mismatch(o, selector, wanted)})" for o in offerings[:5] + ) + raise _refused( + f"No advertised offering matches this selector. Nearest: {nearest}. Adjust the " + "selector to an offering the seller advertises.", + "SUBSCRIPTION_SELECTOR_UNMATCHED", + ) + + # -- the guarded write --------------------------------------------------------------- + + async def _write( + self, build: Callable[[_Principal], Awaitable[_Plan]], *, single: bool = True + ) -> SubscribeResult: + principal = await read_principal(self._client) + for attempt in range(1, self._attempts + 1): + plan = await build(principal) + if plan.configuration is None: + return plan.result(principal, self._readback(principal)) + key = self._key() + outcome = await self._send(principal, plan.configuration, key=key) + if outcome is not None: + readback = ReportingSubscriptionReadback( + configuration_version=getattr(outcome, "configuration_version", None), + subscriptions=tuple( + getattr( + getattr(outcome, "configuration", None), "reporting_subscriptions", None + ) + or () + ), + ) + destinations = tuple( + getattr(getattr(outcome, "configuration", None), "reporting_destinations", None) + or () + ) + self._verify(plan, readback) + return plan.result( + _Principal( + "current", + readback.configuration_version, + principal.principal_kind, + readback.subscriptions, + destinations, + ), + readback, + ) + if attempt < self._attempts: + principal = await read_principal(self._client) + raise _refused( + f"The principal configuration changed during all {self._attempts} attempts; " + "nothing was applied by the last one. Another writer is updating this " + "principal; retry later.", + "SUBSCRIPTION_CONFLICT_EXHAUSTED", + ) + + async def _send( + self, + principal: _Principal, + configuration: Mapping[str, Any], + *, + key: str, + dry_run: bool = False, + ) -> Any | None: + """One attempt; ``None`` on CONFLICT. A transport failure retries the same key once.""" + payload: dict[str, Any] = {"idempotency_key": key, "configuration": dict(configuration)} + if principal.kind == "current" and principal.configuration_version is not None: + payload["expected_configuration_version"] = principal.configuration_version + if principal.principal_kind is not None and principal.kind in {"current", "recognized"}: + payload["expected_principal_kind"] = principal.principal_kind + if dry_run: + payload["dry_run"] = True + request = SyncPrincipalRequest.model_validate(payload) + sync = ( + getattr(self._client, "_sync_principal_unguarded", None) or self._client.sync_principal + ) + try: + result = await sync(request) + except (OSError, TimeoutError): + result = await sync(request) # same key and body: the seller replays + response = getattr(result, "data", None) + if not getattr(result, "success", False) or response is None: + raise ReportingSubscriptionError( + f"sync_principal failed: {getattr(result, 'error', None) or 'no response'}. " + "Retry the same call; the SDK reuses the key only within one attempt." + ) + outcome = response.result + kind = _text(getattr(outcome, "kind", "")) + if kind == "failed": + errors = list(getattr(outcome, "errors", None) or []) + if any(_text(getattr(e, "code", "")) == "CONFLICT" for e in errors): + return None + first = errors[0] if errors else None + recovery = getattr(first, "recovery", None) + raise ReportingSubscriptionError( + f"sync_principal was rejected: {getattr(first, 'message', 'unknown error')}", + code=None if first is None else (_text(getattr(first, "code", "")) or None), + recovery=_text(recovery) or None, # type: ignore[arg-type] + ) + return outcome + + @staticmethod + def _readback(principal: _Principal) -> ReportingSubscriptionReadback: + return ReportingSubscriptionReadback( + configuration_version=principal.configuration_version, + subscriptions=principal.subscriptions, + ) + + @staticmethod + def _verify(plan: _Plan, readback: ReportingSubscriptionReadback) -> None: + """Every submitted subscription reads back at its version, state and content.""" + for sent in plan.verify: + state = readback.get(sent["subscription_id"]) + + def fail(detail: str) -> ReportingSubscriptionError: + return _refused( + f"The seller's readback for subscription {sent['subscription_id']!r} " + f"{detail}. The seller altered a submitted field; contact the seller.", + "SUBSCRIPTION_READBACK_MISMATCH", + readback=readback, + ) + + if state is None or state.subscription_version != sent.get("subscription_version"): + raise fail(f"does not carry version {sent.get('subscription_version')}") + if (sent.get("active") is False) != (_text(state.state) == "inactive"): + raise fail(f"does not reflect active: {sent.get('active')}") + submitted = {k: v for k, v in sent.items() if k != "active"} + stored = {k: v for k, v in _dump(state.configuration).items() if k != "active"} + found = differing_path(submitted, stored) + if found is not None: + raise fail(f"stores a different configuration at {found}") + + +@dataclass +class _Plan: + configuration: dict[str, Any] | None + action: SubscribeAction + version: int + subscription_id: str | None + verify: list[dict[str, Any]] = field(default_factory=list) + offering_ids: tuple[str, ...] = () + + @classmethod + def unchanged(cls, principal: _Principal, wanted: Mapping[str, Any]) -> _Plan: + return cls( + None, + "unchanged", + int(wanted.get("subscription_version") or 1), + str(wanted.get("subscription_id")), + ) + + def result( + self, principal: _Principal, readback: ReportingSubscriptionReadback + ) -> SubscribeResult: + state = ( + None if self.subscription_id is None else principal.subscription(self.subscription_id) + ) + destination = None + target = None if state is None else _value(_value(state, "configuration"), "destination") + if target is not None: + destination_id = _value(target, "destination_id") + destination = next( + ( + d + for d in principal.destinations + if _value(d, "destination_id") == destination_id + ), + None, + ) + return SubscribeResult(self.action, state, self.version, destination, readback) + + +def _mismatch(offering: Any, selector: Mapping[str, Any], wanted: Mapping[str, Any]) -> str: + """The first selector attribute an offering fails, for SUBSCRIPTION_SELECTOR_UNMATCHED.""" + if offering.feed_purpose != selector.get("feed_purpose"): + return f"feed_purpose {offering.feed_purpose}" + if offering.period_duration != selector.get("period_duration"): + return f"period_duration {offering.period_duration}" + if selector.get("required_finality") not in offering.supported_finality: + return f"finality {'/'.join(sorted(offering.supported_finality))}" + if offering.reconciliation_mode != wanted.get("reconciliation_mode", "delivery_only"): + return f"reconciliation_mode {offering.reconciliation_mode}" + wanted_dimensions = set(selector.get("dimensions") or ()) + if not wanted_dimensions <= set(offering.dimensions): + return f"dimensions {'/'.join(sorted(offering.dimensions))}" + return "destination method" + + +__all__ = [ + "PreviewResult", + "ReportingClient", + "SubscribeResult", + "guard_sync_principal", +] diff --git a/tests/test_reporting_subscription_facade.py b/tests/test_reporting_subscription_facade.py new file mode 100644 index 000000000..b8c46e921 --- /dev/null +++ b/tests/test_reporting_subscription_facade.py @@ -0,0 +1,346 @@ +"""The buyer write facade (``client.reporting``) and the raw ``sync_principal`` guard. + +A real in-process seller (``PrincipalService`` plus the subscription engine) sits +behind the test client, so the buyer sees only wire responses. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from adcp.exceptions import is_pre_dispatch_error +from adcp.reporting.subscriptions import ReportingSubscriptionError +from adcp.reporting.subscriptions.facade import ReportingClient, guard_sync_principal +from adcp.types import SyncPrincipalRequest +from adcp.types.core import TaskResult, TaskStatus +from tests.test_reporting_subscription_consumer import SUBSCRIPTION, Client, Seller + +pytestmark = pytest.mark.asyncio + +DESTINATION = { + "pattern": "warehouse_materialization", + "destination_id": "warehouse", + "active": True, + "provider": {"domain": "fixture.example.test"}, + "transport": "fixture-sql", + "location": "project.reporting", + "accepted_verification_profiles": ["canonical_digest"], +} + + +def _subscription(subscription_id: str = "analytics", **changes: Any) -> dict[str, Any]: + value = {k: v for k, v in SUBSCRIPTION.items() if k != "subscription_version"} + value["subscription_id"] = subscription_id + value.update(changes) + return value + + +class Counting(Client): + def __init__(self, seller: Seller) -> None: + super().__init__(seller) + self.writes = 0 + + async def sync_principal(self, request): + self.writes += 1 + return await super().sync_principal(request) + + +def _state(readback, subscription_id): + state = readback.get(subscription_id) + assert state is not None + return state + + +async def test_subscribe_creates_auto_versions_and_keeps_the_others() -> None: + client = Counting(Seller()) + reporting = ReportingClient(client) + created = await reporting.subscribe(_subscription()) + assert (created.action, created.subscription_version) == ("created", 1) + assert created.state.state == "active" + # A second subscription leaves the first untouched. + await reporting.subscribe(_subscription("pacing", scope={"account_ids": ["acct-1"]})) + listed = await reporting.list_subscriptions() + assert {s.subscription_id for s in listed.subscriptions} == {"analytics", "pacing"} + assert all(s.state == "active" for s in listed.subscriptions) + # The same content writes nothing. + writes = client.writes + again = await reporting.subscribe(_subscription()) + assert (again.action, again.subscription_version) == ("unchanged", 1) + assert client.writes == writes + # Changed content bumps the version. + selector = {**SUBSCRIPTION["selector"], "required_finality": "official", "match": "all"} + updated = await reporting.subscribe(_subscription(selector=selector)) + assert (updated.action, updated.subscription_version) == ("updated", 2) + assert _state(await reporting.list_subscriptions(), "pacing").subscription_version == 1 + + +async def test_a_supplied_version_must_exceed_the_current_one_when_content_changes() -> None: + reporting = ReportingClient(Client(Seller())) + await reporting.subscribe(_subscription()) + selector = {**SUBSCRIPTION["selector"], "match": "all"} + with pytest.raises(ReportingSubscriptionError) as refused: + await reporting.subscribe(_subscription(selector=selector, subscription_version=1)) + assert refused.value.code == "INVALID_REQUEST" and is_pre_dispatch_error(refused.value) + + +async def test_unsubscribe_is_the_deactivation_path_and_resubscribe_reactivates() -> None: + reporting = ReportingClient(Client(Seller())) + await reporting.subscribe(_subscription()) + gone = await reporting.unsubscribe("analytics") + assert gone.action == "deactivated" and gone.state.state == "inactive" + back = await reporting.subscribe(_subscription()) + assert (back.action, back.subscription_version) == ("reactivated", 1) + assert back.state.state == "active" + with pytest.raises(ReportingSubscriptionError, match="list_subscriptions"): + await reporting.unsubscribe("missing") + + +async def test_pre_dispatch_checks_send_nothing() -> None: + client = Counting(Seller()) + reporting = ReportingClient(client) + with pytest.raises(ReportingSubscriptionError) as scope: + await reporting.subscribe(_subscription(scope={})) + assert is_pre_dispatch_error(scope.value) and "all_authorized_accounts" in str(scope.value) + with pytest.raises(ReportingSubscriptionError, match="register_destination"): + await reporting.subscribe( + _subscription( + destination={"destination_id": "warehouse", "path_template": "a/{account_id}"} + ) + ) + with pytest.raises(ReportingSubscriptionError, match=r"\{account_id\}"): + await reporting.subscribe( + _subscription(destination={"destination_id": "warehouse", "path_template": "flat"}), + register_destination=DESTINATION, + ) + assert client.writes == 0 + + +async def test_raw_writes_are_guarded_against_deactivation_and_revocation() -> None: + client = Client(Seller()) + reporting = ReportingClient(client) + await reporting.subscribe(_subscription()) + await reporting.subscribe(_subscription("pacing", scope={"account_ids": ["acct-1"]})) + raw = SyncPrincipalRequest.model_validate( + { + "idempotency_key": "raw-write-0000000001", + "configuration": {"reporting_subscriptions": [{**SUBSCRIPTION}]}, + } + ) + with pytest.raises(ReportingSubscriptionError) as refused: + await guard_sync_principal(client, raw) + error = refused.value + assert error.code == "SUBSCRIPTION_WOULD_DEACTIVATE" and is_pre_dispatch_error(error) + assert error.would_deactivate == ("pacing",) and error.recovery == "correctable" + await guard_sync_principal(client, raw, allow_deactivation=True) + # An explicit ``active: false`` is a deliberate deactivation. + pacing = ( + (await reporting.list_subscriptions()) + .get("pacing") + .configuration.model_dump(mode="json", exclude_none=True) + ) + explicit = SyncPrincipalRequest.model_validate( + { + "idempotency_key": "raw-write-0000000002", + "configuration": { + "reporting_subscriptions": [{**SUBSCRIPTION}, {**pacing, "active": False}] + }, + } + ) + await guard_sync_principal(client, explicit) + + +async def test_omitting_a_destination_is_refused_unless_allowed() -> None: + client = Client(Seller()) + reporting = ReportingClient(client) + await reporting.subscribe( + _subscription( + destination={"destination_id": "warehouse", "path_template": "a/{account_id}"} + ), + register_destination=DESTINATION, + ) + raw = SyncPrincipalRequest.model_validate( + {"idempotency_key": "raw-write-0000000003", "configuration": {"reporting_destinations": []}} + ) + with pytest.raises(ReportingSubscriptionError) as refused: + await guard_sync_principal(client, raw) + assert refused.value.code == "DESTINATION_WOULD_REVOKE" + await guard_sync_principal(client, raw, allow_revocation=True) + + +async def test_the_adcp_client_guards_raw_sync_principal() -> None: + from adcp.client import ADCPClient + from adcp.types.core import AgentConfig, Protocol + + adcp_client = ADCPClient( + AgentConfig(id="seller", agent_uri="https://seller.example/mcp", protocol=Protocol.MCP) + ) + seller = Client(Seller()) + await ReportingClient(seller).subscribe(_subscription()) + await ReportingClient(seller).subscribe( + _subscription("pacing", scope={"account_ids": ["acct-1"]}) + ) + adcp_client.get_principal = seller.get_principal # type: ignore[method-assign] + unguarded = AsyncMock() + adcp_client._sync_principal_unguarded = unguarded # type: ignore[method-assign] + raw = SyncPrincipalRequest.model_validate( + { + "idempotency_key": "raw-write-0000000004", + "configuration": {"reporting_subscriptions": [{**SUBSCRIPTION}]}, + } + ) + with pytest.raises(ReportingSubscriptionError): + await adcp_client.sync_principal(raw) + unguarded.assert_not_awaited() + await adcp_client.sync_principal(raw, allow_deactivation=True) + unguarded.assert_awaited_once() + assert isinstance(adcp_client.reporting, ReportingClient) + + +async def test_replace_subscriptions_needs_explicit_flags() -> None: + reporting = ReportingClient(Client(Seller())) + await reporting.subscribe(_subscription()) + with pytest.raises(ReportingSubscriptionError, match="deactivate_all"): + await reporting.replace_subscriptions([]) + other = { + **_subscription("pacing", scope={"account_ids": ["acct-1"]}), + "subscription_version": 1, + } + with pytest.raises(ReportingSubscriptionError) as refused: + await reporting.replace_subscriptions([other]) + assert refused.value.code == "SUBSCRIPTION_WOULD_DEACTIVATE" + readback = await reporting.replace_subscriptions([other], deactivate_omitted=True) + assert _state(readback, "analytics").state == "inactive" + + +class Interleaving(Client): + """Another writer changes the principal before each of this client's writes.""" + + def __init__(self, seller: Seller, times: int) -> None: + super().__init__(seller) + self.times, self.keys = times, [] + + async def sync_principal(self, request): + self.keys.append(request.idempotency_key) + if self.times > 0: + self.times -= 1 + other = Client(self.seller) + await ReportingClient(other).subscribe( + _subscription(f"other-{self.times}", scope={"account_ids": [f"acct-{self.times}"]}) + ) + return await super().sync_principal(request) + + +async def test_a_conflict_retries_with_a_fresh_key_and_then_gives_up() -> None: + seller = Seller() + # A configured principal, so each write carries expected_configuration_version. + await ReportingClient(Client(seller)).subscribe( + _subscription("seed", scope={"account_ids": ["acct-seed"]}) + ) + client = Interleaving(seller, times=1) + result = await ReportingClient(client).subscribe(_subscription()) + assert result.action == "created" and len(set(client.keys)) == 2 + # The interleaved writer's subscription survived the re-merge. + assert _state(result.readback, "other-0").state == "active" + stuck_seller = Seller() + await ReportingClient(Client(stuck_seller)).subscribe( + _subscription("seed", scope={"account_ids": ["acct-seed"]}) + ) + stuck = Interleaving(stuck_seller, times=10) + with pytest.raises(ReportingSubscriptionError) as exhausted: + await ReportingClient(stuck, max_attempts=2).subscribe(_subscription()) + assert exhausted.value.code == "SUBSCRIPTION_CONFLICT_EXHAUSTED" + assert exhausted.value.recovery == "transient" and not is_pre_dispatch_error(exhausted.value) + + +class Altering(Client): + async def sync_principal(self, request): + result = await super().sync_principal(request) + state = result.data.result.configuration.reporting_subscriptions[0] + state.subscription_version = 99 + return result + + +async def test_a_readback_mismatch_is_terminal() -> None: + with pytest.raises(ReportingSubscriptionError) as mismatch: + await ReportingClient(Altering(Seller())).subscribe(_subscription()) + assert mismatch.value.code == "SUBSCRIPTION_READBACK_MISMATCH" + assert mismatch.value.recovery == "terminal" and mismatch.value.readback is not None + + +class Roster(Client): + def __init__(self, seller: Seller, accounts: list[dict[str, Any]], offerings=None) -> None: + super().__init__(seller) + self.accounts, self.offerings = accounts, offerings + + async def list_accounts(self, request): + from types import SimpleNamespace + + data = SimpleNamespace(accounts=self.accounts, pagination=None) + return TaskResult.model_construct(status=TaskStatus.COMPLETED, data=data, success=True) + + async def fetch_capabilities(self): + from types import SimpleNamespace + + media_buy = {"reporting_delivery": {"offerings": self.offerings or []}} + return SimpleNamespace( + model_dump=lambda **_: {"media_buy": media_buy} if self.offerings else {} + ) + + +def _account(account_id: str, operator: str, configs=()) -> dict[str, Any]: + return { + "account_id": account_id, + "name": account_id, + "status": "active", + "operator": operator, + "reporting_delivery_configs": list(configs), + } + + +async def test_roster_checks_refuse_an_id_clash_and_cross_operator_scope() -> None: + clash = Roster( + Seller(), + [ + _account( + "acct-1", "op-a.example", [{"configuration": {"delivery_config_id": "analytics"}}] + ) + ], + ) + with pytest.raises(ReportingSubscriptionError) as refused: + await ReportingClient(clash).subscribe(_subscription()) + assert refused.value.code == "SUBSCRIPTION_ID_CLASH" and is_pre_dispatch_error(refused.value) + two = Roster(Seller(), [_account("a", "op-a.example"), _account("b", "op-b.example")]) + with pytest.raises(ReportingSubscriptionError) as scoped: + await ReportingClient(two).subscribe(_subscription()) + assert scoped.value.code == "SUBSCRIPTION_OPERATOR_SCOPE_REQUIRED" + created = await ReportingClient(two).subscribe(_subscription(), acknowledge_cross_operator=True) + assert created.action == "created" + + +def _offering(offering_id: str, period: str, dimensions: list[str]) -> dict[str, Any]: + return { + "offering_id": offering_id, + "report_definition_id": "def-pkg", + "reporting_profile": {"id": "delivery"}, + "feed_purpose": "analytics", + "schedule": {"period_duration": period, "alignment": "utc", "delivery_sla": "PT6H"}, + "supported_finality": ["official"], + "reconciliation_mode": "delivery_only", + "standard_dimensions": dimensions, + } + + +async def test_a_selector_no_advertised_offering_matches_is_refused_before_dispatch() -> None: + hourly = Roster(Seller(), [], offerings=[_offering("hourly", "PT1H", ["package"])]) + with pytest.raises(ReportingSubscriptionError) as unmatched: + await ReportingClient(hourly).subscribe(_subscription()) + error = unmatched.value + assert error.code == "SUBSCRIPTION_SELECTOR_UNMATCHED" and is_pre_dispatch_error(error) + assert "hourly (period_duration PT1H)" in str(error) + daily = Roster(Seller(), [], offerings=[_offering("pkg", "P1D", ["package"])]) + preview = await ReportingClient(daily).preview(_subscription()) + assert (preview.would_be, preview.subscription_version) == ("created", 1) + assert preview.offering_ids == ("pkg",)