diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index df9ec9fd4..581f2f8ba 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -318,6 +318,82 @@ than `configure()`. Production workers belong to `start()`/`close()`; the Core [production guide](reporting-production.md) for account admission and provider contracts, and [the wiring example](../examples/reporting_service_production.py). +## Reporting subscriptions (AdCP 3.3, experimental) + +`media_buy.reporting_subscriptions` lets a caller register reporting once on its +principal through `sync_principal`, rather than once per account through +`sync_accounts`. `adcp.reporting.subscriptions` holds both halves. + +**Buyer.** `expected_subscription_periods(...)` is the coverage rule as a pure +function over plain data. Pass it the `get_principal` subscription readback, +the buys you can see (`SubscriptionBuy`, with earlier flights when a buy was +extended or reinstated), the offerings with the seller's dimension mapping +(`SubscriptionOffering`), the account timezone, the account's grant history, +and the `subscription_coverage[]` exceptions and gaps. It returns +`ExpectedSubscriptionPeriod` values, which are `ExpectedReportingPeriod`s, so +they feed `evaluate_reporting_ledger` and the consumer-status planner as they +are. + +**Seller.** Opt in on the principal runtime and link it to the engine. Without +`reporting_subscriptions=`, `PrincipalService` rejects the section with +`UNSUPPORTED_FEATURE`, as the spec requires for an unsupported section. +`replace_account_configs: true` is also rejected with `UNSUPPORTED_FEATURE` +until the engine can deactivate account-level configurations. + +```python +from adcp.reporting.subscriptions import ( + ReportingSubscriptionEngine, + ReportingSubscriptionSupport, + SubscriptionOffering, +) +from adcp.server import PrincipalService + +principals = PrincipalService(emit_change=emit_principal_changed) +engine = ReportingSubscriptionEngine( + principals=principals, + host=my_host, # list_authorized_accounts, account_buys[, account_delivery_configs] + offerings=[ + SubscriptionOffering.from_offering(daily_offering), # reads standard_dimensions + ], + support=ReportingSubscriptionSupport(max_accounts=500), +) +principals.set_reporting_subscriptions( + engine.principal_options(delivery_config_ids=caller_config_ids) +) +``` + +The host supplies only what it alone knows: the accounts a principal may read +(`list_authorized_accounts`, paged), each with its IANA `timezone` and the +`authorized_at` that `list_accounts` serves (the engine never substitutes +either); each account's buys with their products' +`reporting_delivery_offering_ids` (`account_buys`); and optionally the caller's +own active `sync_accounts` configurations (`account_delivery_configs`), which +take precedence. Give each configuration its `active_from` so the `overridden` +exception opens at that instant; otherwise run `engine.reconcile(principal)` +synchronously with every `sync_accounts` write. Call `engine.reconcile` on a +schedule and after anything changes, and `engine.authorization_changed(...)` +when you learn of a grant or revocation directly. A subscription that cannot +be evaluated (for example an account with no timezone yet) is isolated and +listed in `engine.evaluation_errors`; the others carry on. `subscription_coverage` +answers from the engine's last-reconciled authority, so re-check authority +before serving it. Use `render_path_template(...)` to expand a destination +`path_template`, and `PrincipalService.reporting_subscription_ids(...)` to make +`sync_accounts` reject a `delivery_config_id` that equals a subscription id. The engine records exceptions as cutover +intervals, keeps `accounts_with_exceptions[]` and `accounts_unauthorized[]` +current in `get_principal` (each change fires `principal.changed` with reason +`reporting_subscription_changed`), serves `subscription_coverage(...)` for +`get_reporting_status`, and computes `owed_periods(...)` with the same function +the buyer runs. + +Under `max_accounts`, accounts already served keep their slots and the rest are +filled in `account_id` byte order; the remainder get a `capacity` exception. + +`subscription_capabilities(...)` and `advertise_reporting_subscriptions(...)` +project `configuration_tasks`, the deprecated `configuration_task`, the +`subscriptions` block, the feature id and the principal section. The engine +does not yet write ledger obligations, so do not advertise the capability +until obligations are materialized for the periods it owes. + ## Advanced composition `postgres_reliable_reporting_service()` (from `adcp_reporting_storage`) without diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index 7eec11a06..bd2bdc82c 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -12,6 +12,11 @@ (``exact`` / ``finest`` / ``all``, ties to the lowest ``offering_id``). * :func:`subscription_capabilities` -- the capability fields a seller that serves subscriptions must advertise. +* :func:`apply_reporting_subscriptions` -- validation and versioning of the + ``sync_principal`` section, used by ``PrincipalService`` when it is given + ``reporting_subscriptions=`` options. +* :class:`ReportingSubscriptionEngine` -- the seller runtime: host hooks, + lost authority, exceptions, ``subscription_coverage[]`` and owed periods. """ from __future__ import annotations @@ -39,6 +44,22 @@ subscription_exceptions, subscription_version_windows, ) +from adcp.reporting.subscriptions.engine import ( + AccountDeliveryConfig, + AuthorizedAccount, + AuthorizedAccountPage, + InMemoryReportingSubscriptionStore, + ReportingSubscriptionDeliveryHost, + ReportingSubscriptionEngine, + ReportingSubscriptionHost, + ReportingSubscriptionPrecedenceHost, +) +from adcp.reporting.subscriptions.registry import ( + ReportingSubscriptionError, + apply_reporting_subscriptions, + refresh_destination_states, + render_path_template, +) from adcp.reporting.subscriptions.resolution import ( STANDARD_REPORTING_DIMENSIONS, SubscriptionDestination, @@ -57,6 +78,18 @@ ) __all__ = [ + "AccountDeliveryConfig", + "AuthorizedAccount", + "AuthorizedAccountPage", + "InMemoryReportingSubscriptionStore", + "ReportingSubscriptionDeliveryHost", + "ReportingSubscriptionEngine", + "ReportingSubscriptionError", + "ReportingSubscriptionHost", + "ReportingSubscriptionPrecedenceHost", + "apply_reporting_subscriptions", + "refresh_destination_states", + "render_path_template", "REPORTING_SUBSCRIPTIONS_FEATURE", "REPORTING_SUBSCRIPTIONS_SECTION", "STANDARD_REPORTING_DIMENSIONS", diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py new file mode 100644 index 000000000..cf1b9232e --- /dev/null +++ b/src/adcp/reporting/subscriptions/engine.py @@ -0,0 +1,1196 @@ +"""Seller runtime for principal-level reporting subscriptions (experimental). + +:class:`ReportingSubscriptionEngine` turns accepted subscriptions into what the +seller owes, using only facts the host alone knows: + +* :meth:`ReportingSubscriptionHost.list_authorized_accounts`: the accounts a + principal may read, paged. The SDK never infers authority. +* :meth:`ReportingSubscriptionEngine.authorization_changed` (or + :meth:`~ReportingSubscriptionEngine.reconcile`, which diffs the host's + answer) records grants and losses, so the SDK can stamp ``unauthorized_at`` + and re-grants. +* :meth:`ReportingSubscriptionHost.account_buys`: buy facts per account + (:class:`~adcp.reporting.subscriptions.SubscriptionBuy`), including each + package's ``reporting_delivery_offering_ids`` and earlier flights. +* Optionally :meth:`ReportingSubscriptionHost.account_delivery_configs`: the + caller's own active ``sync_accounts`` configurations, for precedence. + +From those the engine records exceptions as cutover intervals (precedence, +``max_accounts`` capacity, ``exclude_account_ids``, unsupported offerings), +tracks lost authority, serves ``subscription_coverage[]`` for +``get_reporting_status``, keeps ``accounts_with_exceptions[]`` and +``accounts_unauthorized[]`` current in the ``get_principal`` readback, and +computes owed periods with the same +:func:`~adcp.reporting.subscriptions.expected_subscription_periods` a buyer +runs, so both sides agree period by period. + +``max_accounts`` order + Among accounts eligible to be served (in scope, authorized, not excluded + or overridden), accounts already served keep their slots; the remaining + slots go to the others in ``account_id`` byte order. Accounts beyond the + limit get a ``capacity`` exception. A newly granted account never + displaces one being served. This matches the TypeScript SDK. +""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass, field, replace +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING, Any, Protocol + +from adcp.reporting.subscriptions._clock import PeriodCalendar, iso_z, require_zone, utc_instant +from adcp.reporting.subscriptions.capabilities import ReportingSubscriptionSupport +from adcp.reporting.subscriptions.coverage import ( + AccountTimezoneChange, + ExpectedSubscriptionPeriod, + SubscriptionAuthorization, + SubscriptionBuy, + SubscriptionException, + SubscriptionVersionWindow, + expected_subscription_periods, + subscription_covers_account, + subscription_version_windows, +) +from adcp.reporting.subscriptions.resolution import ( + SubscriptionDestination, + SubscriptionOffering, + SubscriptionSelector, + resolve_subscription_offerings, +) +from adcp.reporting.subscriptions.wire import project_subscription_coverage +from adcp.types.domains.core.reporting_subscription_coverage import ( + ReportingSubscriptionCoverage, +) +from adcp.types.domains.core.reporting_subscription_state import ReportingSubscriptionState + +logger = logging.getLogger(__name__) + +if TYPE_CHECKING: + from adcp.server.principal import ( + DeliveryConfigIdsHook, + PrincipalIdentity, + PrincipalService, + ReportingSubscriptionOptions, + ) + + +# -- host facts ---------------------------------------------------------------- + + +@dataclass(frozen=True) +class AuthorizedAccount: + """One account the principal may read, as the host answers it. + + ``timezone`` is the account's IANA zone; the engine never substitutes one. + ``authorized_at`` is the start of the caller's current authorization chain, + exactly what ``list_accounts`` serves as ``authorization.authorized_at`` + (sellers advertising subscriptions MUST populate it), so the buyer and the + engine start coverage at the same instant. + """ + + account_id: str + timezone: str + authorized_at: datetime + operator: str | None = None + + def __post_init__(self) -> None: + require_zone(self.timezone, field="AuthorizedAccount.timezone") + object.__setattr__( + self, "authorized_at", utc_instant(self.authorized_at, field="authorized_at") + ) + + +@dataclass(frozen=True) +class AuthorizedAccountPage: + accounts: tuple[AuthorizedAccount, ...] + next_cursor: str | None = None + + +@dataclass(frozen=True) +class AccountDeliveryConfig: + """The caller's own active ``sync_accounts`` ``reporting_delivery_configs`` entry. + + ``active_from`` is when it took effect. The ``overridden`` exception opens + at the cutover at that instant rather than at the reconcile that first sees + it, so the two registrations never both owe a period. Without it, run + :meth:`ReportingSubscriptionEngine.reconcile` synchronously with every + ``sync_accounts`` write. + """ + + delivery_config_id: str + feed_purpose: str + active_from: datetime | None = None + + +class ReportingSubscriptionHost(Protocol): + """Facts only the seller's host knows. The SDK never infers them.""" + + async def list_authorized_accounts( + self, principal: PrincipalIdentity, *, cursor: str | None + ) -> AuthorizedAccountPage: ... + + async def account_buys( + self, principal: PrincipalIdentity, account_id: str + ) -> Sequence[SubscriptionBuy]: ... + + +class ReportingSubscriptionDeliveryHost(Protocol): + """Optional: delivery facts, for an exact ``withheld_period_count``. + + Without it the engine counts the owed periods not yet due + (``period_end + delivery_sla``) when authority ended, as the TypeScript SDK + does. With it, a period is withheld when it ended at or before + ``unauthorized_at`` and was not delivered by then. + """ + + async def is_period_delivered( + self, + principal: PrincipalIdentity, + account_id: str, + period: ExpectedSubscriptionPeriod, + *, + by: datetime, + ) -> bool: ... + + +class ReportingSubscriptionPrecedenceHost(Protocol): + """Optional: the caller's own account-level configurations, for precedence.""" + + async def account_delivery_configs( + self, principal: PrincipalIdentity, account_id: str + ) -> Sequence[AccountDeliveryConfig]: ... + + +# -- engine state -------------------------------------------------------------- + + +@dataclass +class _Account: + #: ``None`` while the host has not said; account-timezone offerings then + #: fail for this account (isolated per subscription) rather than use UTC. + timezone: str | None + operator: str | None + grants: list[SubscriptionAuthorization] = field(default_factory=list) + timezone_changes: list[AccountTimezoneChange] = field(default_factory=list) + + def set_timezone(self, timezone: str | None, at: datetime) -> None: + """Record a zone; a change away from an unknown zone is not a cutover.""" + if timezone is None or timezone == self.timezone: + return + if self.timezone is not None: + self.timezone_changes.append( + AccountTimezoneChange(at, timezone, previous_timezone=self.timezone) + ) + self.timezone = timezone + + @property + def authorized(self) -> bool: + return bool(self.grants) and self.grants[-1].unauthorized_at is None + + +@dataclass +class _Exclusion: + reason: str + from_: datetime + until: datetime | None = None + media_buy_id: str | None = None + package_ids: tuple[str, ...] = () + overridden_by: tuple[str, str] | None = None + #: When the cause took effect, if known; the exception opens at its cutover. + effective_at: datetime | None = None + + @property + def key(self) -> tuple[Any, ...]: + return (self.reason, self.media_buy_id, self.package_ids, self.overridden_by) + + def wire(self) -> dict[str, Any]: + payload: dict[str, Any] = {"reason": self.reason, "from": iso_z(self.from_)} + if self.media_buy_id is not None: + payload["media_buy_id"] = self.media_buy_id + if self.until is not None: + payload["until"] = iso_z(self.until) + if self.package_ids: + payload["package_ids"] = list(self.package_ids) + if self.overridden_by is not None: + payload["overridden_by"] = {self.overridden_by[0]: self.overridden_by[1]} + return payload + + +@dataclass +class _PrincipalState: + accounts: dict[str, _Account] = field(default_factory=dict) + #: (subscription_id, version) -> configuration + versions: dict[tuple[str, int], dict[str, Any]] = field(default_factory=dict) + #: subscription_id -> accounts it ever covered + covered: dict[str, set[str]] = field(default_factory=dict) + #: subscription_id -> accounts holding a max_accounts slot + served: dict[str, set[str]] = field(default_factory=dict) + #: (subscription_id, account_id) -> exclusions, open and closed + exclusions: dict[tuple[str, str], list[_Exclusion]] = field(default_factory=dict) + #: (subscription_id, version) -> the named destination's configuration, + #: snapshotted at acceptance so a later removal never breaks old versions + destinations: dict[tuple[str, int], Any] = field(default_factory=dict) + #: (subscription_id, account_id, unauthorized_at) -> withheld count from the + #: host's delivery facts; delivery by a past instant never changes + withheld: dict[tuple[str, str, datetime], int] = field(default_factory=dict) + + +class InMemoryReportingSubscriptionStore: + """Process-local engine state. Durable storage is a later slice.""" + + def __init__(self) -> None: + self.principals: dict[str, _PrincipalState] = {} + self.lock = asyncio.Lock() + + def principal(self, subject: str) -> _PrincipalState: + return self.principals.setdefault(subject, _PrincipalState()) + + +def _dump(value: Any) -> dict[str, Any]: + if hasattr(value, "model_dump"): + dumped = value.model_dump(mode="json", exclude_none=True) + assert isinstance(dumped, dict) + return dumped + return dict(value) + + +def _active(state: Mapping[str, Any]) -> bool: + return state.get("state") != "inactive" + + +class ReportingSubscriptionEngine: + """Owed periods, exceptions and lost authority for accepted subscriptions.""" + + def __init__( + self, + *, + principals: PrincipalService, + host: ReportingSubscriptionHost, + offerings: Sequence[SubscriptionOffering], + support: ReportingSubscriptionSupport | None = None, + status_retention_days: int = 400, + store: InMemoryReportingSubscriptionStore | None = None, + clock: Callable[[], datetime] | None = None, + ) -> None: + ids = [item.offering_id for item in offerings] + if len(set(ids)) != len(ids): + raise ValueError("offering_id values must be unique") + for offering in offerings: + # Refuse what would otherwise fail mid-evaluation for every + # subscription that resolves the offering. + if offering.alignment == "source_timezone" and offering.period_timezone is None: + raise ValueError( + f"offering {offering.offering_id!r} resolves its source timezone per " + "account; register it with the resolved period_timezone" + ) + if offering.alignment != "account_timezone": + offering.calendar(None) + else: + offering.calendar("UTC") # validates the duration for a local alignment + self._principals = principals + self._host = host + self._offerings = tuple(offerings) + self._support = support or ReportingSubscriptionSupport() + self._retention = timedelta(days=status_retention_days) + self._retention_days = status_retention_days + self._store = store or InMemoryReportingSubscriptionStore() + self._clock = clock or (lambda: datetime.now(timezone.utc)) + self._errors: dict[tuple[str, str], Exception] = {} + + @property + def evaluation_errors(self) -> Mapping[tuple[str, str], Exception]: + """``(principal subject, subscription_id)`` -> the last evaluation failure. + + One subscription that cannot be evaluated never stops the others; its + failure stays here until a later reconcile succeeds for it. + """ + return dict(self._errors) + + @property + def support(self) -> ReportingSubscriptionSupport: + return self._support + + def principal_options( + self, *, delivery_config_ids: DeliveryConfigIdsHook | None = None + ) -> ReportingSubscriptionOptions: + """Options that link a ``PrincipalService`` to this engine.""" + from adcp.server.principal import ReportingSubscriptionOptions + + return ReportingSubscriptionOptions( + delivery_config_ids=delivery_config_ids, + record_versions=self.record_versions, + clock=self._clock, + ) + + def _now(self, now: datetime | None) -> datetime: + return utc_instant(now if now is not None else self._clock(), field="now") + + # -- recording ------------------------------------------------------------- + + async def record_versions( + self, identity: PrincipalIdentity, states: Sequence[ReportingSubscriptionState] + ) -> None: + """Keep every version's configuration; the readback keeps only the current one. + + In-place reactivations need no engine state: the readback's + ``inactive_intervals`` (adcp#8585) carry them. Each version's named + destination is snapshotted, so removing it later never breaks that + version's coverage. + """ + destinations = await self._destinations(identity) + async with self._store.lock: + principal = self._store.principal(identity.subject) + for state in states: + payload = _dump(state) + key = (str(payload["subscription_id"]), int(payload["subscription_version"])) + configuration = dict(payload["configuration"]) + principal.versions.setdefault(key, configuration) + destination = configuration.get("destination") + if destination is not None and key not in principal.destinations: + target = destinations.get(destination["destination_id"]) + if target is not None: + principal.destinations[key] = target + + async def authorization_changed( + self, + identity: PrincipalIdentity, + account_id: str, + *, + authorized: bool, + at: datetime, + timezone: str | None = None, + operator: str | None = None, + ) -> None: + """Record a grant or a loss of authority at ``at``.""" + instant = utc_instant(at, field="at") + async with self._store.lock: + principal = self._store.principal(identity.subject) + account = principal.accounts.get(account_id) + if account is None: + if not authorized: + return + if timezone is not None: + require_zone(timezone, field="timezone") + account = principal.accounts[account_id] = _Account( + timezone=timezone, operator=operator + ) + self._apply_grant(account, authorized=authorized, at=instant) + account.set_timezone(timezone, instant) + if operator is not None: + account.operator = operator + + @staticmethod + def _apply_grant(account: _Account, *, authorized: bool, at: datetime) -> None: + if authorized and not account.authorized: + if account.grants and account.grants[-1].unauthorized_at is not None: + at = max(at, account.grants[-1].unauthorized_at) + account.grants.append(SubscriptionAuthorization(at)) + elif not authorized and account.authorized: + last = account.grants[-1] + account.grants[-1] = SubscriptionAuthorization( + last.authorized_at, max(at, last.authorized_at + timedelta(microseconds=1)) + ) + + async def _list_accounts(self, identity: PrincipalIdentity) -> list[AuthorizedAccount]: + accounts: list[AuthorizedAccount] = [] + cursor: str | None = None + while True: + page = await self._host.list_authorized_accounts(identity, cursor=cursor) + accounts.extend(page.accounts) + if not page.next_cursor: + return accounts + cursor = page.next_cursor + + # -- reconcile ------------------------------------------------------------- + + async def reconcile(self, identity: PrincipalIdentity, *, now: datetime | None = None) -> None: + """Diff the host's authorized accounts, then re-evaluate every exception. + + Call it on a schedule and whenever the host knows something changed + (a grant, a ``sync_accounts`` write, a new buy or product offering). + Exceptions open and close at the first boundary at or after ``now``, + the cutover instant, and each change to an account's exceptions or + authority fires ``principal.changed`` with reason + ``reporting_subscription_changed``. + """ + instant = self._now(now) + listed = {item.account_id: item for item in await self._list_accounts(identity)} + async with self._store.lock: + principal = self._store.principal(identity.subject) + for account_id, item in listed.items(): + account = principal.accounts.get(account_id) + if account is None: + account = principal.accounts[account_id] = _Account( + timezone=item.timezone, operator=item.operator + ) + if not account.authorized: + lost = account.grants[-1].unauthorized_at if account.grants else None + reported = item.authorized_at + # adcp#8585: authorized_at is the chain start. A later one + # is a new chain starting then; an unchanged one is a + # re-grant within retention, which the wire does not + # date, so it is stamped when this reconcile observes it. + if lost is None or reported > lost: + granted = min(reported, instant) + else: + granted = instant + self._apply_grant(account, authorized=True, at=granted) + account.set_timezone(item.timezone, instant) + account.operator = item.operator + for account_id, account in principal.accounts.items(): + if account_id not in listed and account.authorized: + self._apply_grant(account, authorized=False, at=instant) + await self._evaluate(identity, instant) + + async def _evaluate(self, identity: PrincipalIdentity, now: datetime) -> None: + states = [_dump(item) for item in await self._principals.reporting_subscriptions(identity)] + destinations = await self._destinations(identity) + precedence = getattr(self._host, "account_delivery_configs", None) + async with self._store.lock: + principal = self._store.principal(identity.subject) + accounts = dict(principal.accounts) + covered_ids = set().union(*principal.covered.values()) if principal.covered else set() + buys: dict[str, Sequence[SubscriptionBuy]] = {} + configs: dict[str, Sequence[AccountDeliveryConfig]] = {} + for account_id, account in accounts.items(): + if account.authorized or account_id in covered_ids: + buys[account_id] = await self._host.account_buys(identity, account_id) + if account.authorized and precedence is not None: + configs[account_id] = await precedence(identity, account_id) + + updates: list[tuple[str, list[dict[str, Any]], list[dict[str, Any]]]] = [] + async with self._store.lock: + principal = self._store.principal(identity.subject) + for state in states: + subscription_id = str(state["subscription_id"]) + try: + updates.append( + self._evaluate_subscription( + principal, state, states, accounts, buys, configs, destinations, now + ) + ) + except Exception as error: # noqa: BLE001 - isolate one subscription + self._errors[(identity.subject, subscription_id)] = error + logger.warning( + "reporting subscription %s could not be evaluated: %s", + subscription_id, + error, + ) + else: + self._errors.pop((identity.subject, subscription_id), None) + delivered = getattr(self._host, "is_period_delivered", None) + for subscription_id, _with_exceptions, unauthorized in updates: + for lost_entry in unauthorized: + ended = lost_entry.pop("_ended") + if delivered is None: + continue + lost = datetime.fromisoformat(lost_entry["unauthorized_at"].replace("Z", "+00:00")) + cache_key = (subscription_id, lost_entry["account_id"], lost) + async with self._store.lock: + cached = self._store.principal(identity.subject).withheld.get(cache_key) + if cached is None: + # Delivery by a past instant never changes: ask once per loss. + cached = 0 + for period in ended: + if not await delivered(identity, lost_entry["account_id"], period, by=lost): + cached += 1 + async with self._store.lock: + self._store.principal(identity.subject).withheld[cache_key] = cached + lost_entry["withheld_period_count"] = cached + for subscription_id, with_exceptions, unauthorized in updates: + await self._principals.update_subscription_status( + identity, + subscription_id, + accounts_with_exceptions=with_exceptions, + accounts_unauthorized=unauthorized, + changed_at=now, + ) + + def _evaluate_subscription( + self, + principal: _PrincipalState, + state: Mapping[str, Any], + states: Sequence[Mapping[str, Any]], + accounts: Mapping[str, _Account], + buys: Mapping[str, Sequence[SubscriptionBuy]], + configs: Mapping[str, Sequence[AccountDeliveryConfig]], + destinations: Mapping[str, Any], + now: datetime, + ) -> tuple[str, list[dict[str, Any]], list[dict[str, Any]]]: + subscription_id = str(state["subscription_id"]) + desired = self._desired_exclusions( + state, + states, + accounts, + buys, + configs, + destinations, + now, + served=principal.served.setdefault(subscription_id, set()), + ) + covered = principal.covered.setdefault(subscription_id, set()) + for account_id, wanted in desired.items(): + # Covered: served, with no account-level exception. + if not any(item.media_buy_id is None for item in wanted): + covered.add(account_id) + # Surface an account the schedules cannot be derived for (no + # timezone yet) now, rather than at the first owed_periods. + self._schedules( + state, accounts[account_id], buys.get(account_id, ()), destinations, now + ) + self._sync_exclusions( + principal, + subscription_id, + account_id, + wanted, + state, + accounts, + buys.get(account_id, ()), + destinations, + now, + ) + # Exceptions close only when their cause ends. While the caller has + # lost authority for an account they stay as they were: closing them + # would owe the revoked interval again on a backfilled re-grant. + for (sub_id, account_id), entries in principal.exclusions.items(): + if sub_id != subscription_id or account_id in desired: + continue + account = accounts.get(account_id) + if account is not None and not account.authorized: + continue + for entry in entries: + if entry.until is None: + entry.until = self._cutover( + state, account, buys.get(account_id, ()), destinations, now + ) + return ( + subscription_id, + self._accounts_with_exceptions(principal, subscription_id, now), + self._accounts_unauthorized(principal, state, accounts, buys, destinations, now), + ) + + async def _destinations(self, identity: PrincipalIdentity) -> dict[str, Any]: + return { + item.destination_id: item.configuration + for item in await self._principals.reporting_destinations(identity) + } + + def _desired_exclusions( + self, + state: Mapping[str, Any], + states: Sequence[Mapping[str, Any]], + accounts: Mapping[str, _Account], + buys: Mapping[str, Sequence[SubscriptionBuy]], + configs: Mapping[str, Sequence[AccountDeliveryConfig]], + destinations: Mapping[str, Any], + now: datetime, + *, + served: set[str], + ) -> dict[str, list[_Exclusion]]: + """Per account the scope includes: the exclusions that should be open now. + + ``served`` holds the accounts with a ``max_accounts`` slot and is + updated in place. + """ + result: dict[str, list[_Exclusion]] = {} + if not _active(state): + return result + configuration = state["configuration"] + scope = configuration["scope"] + feed_purpose = configuration["selector"]["feed_purpose"] + excluded_ids = set(configuration.get("exclude_account_ids") or ()) + eligible: list[tuple[int, bytes, str]] = [] + for account_id, account in accounts.items(): + if not account.authorized: + continue + if account_id in excluded_ids: + if _operator_selected(configuration, account): + result[account_id] = [_Exclusion("excluded_by_buyer", now)] + continue + if not _scope_selects(configuration, account_id, account): + continue + overriding = _override( + configuration, + scope, + feed_purpose, + account_id, + account, + states, + configs.get(account_id, ()), + ) + if overriding is not None: + kind, identifier, effective_at = overriding + result[account_id] = [ + _Exclusion( + "overridden", + now, + overridden_by=(kind, identifier), + effective_at=effective_at, + ) + ] + continue + result[account_id] = [] + eligible.append( + (0 if account_id in served else 1, account_id.encode("utf-8"), account_id) + ) + limit = self._support.max_accounts + ranked = [account_id for _held, _key, account_id in sorted(eligible)] + kept = ranked if limit is None else ranked[:limit] + served.clear() + served.update(kept) + for account_id in ranked[len(kept) :]: + result[account_id] = [_Exclusion("capacity", now)] + for account_id, entries in result.items(): + if entries: + continue + account = accounts[account_id] + entries.extend( + self._unsupported(state, account, buys.get(account_id, ()), destinations, now) + ) + return result + + def _unsupported( + self, + state: Mapping[str, Any], + account: _Account, + buys: Sequence[SubscriptionBuy], + destinations: Mapping[str, Any], + now: datetime, + ) -> list[_Exclusion]: + selector = SubscriptionSelector.from_subscription(state["configuration"]) + destination = _destination_for(selector, destinations) + entries: list[_Exclusion] = [] + for buy in sorted(buys, key=lambda item: item.media_buy_id.encode("utf-8")): + if buy.confirmed_at is None: + continue + unresolved: list[str] = [] + products = buy.product_offering_ids() + for package_id, offering_ids in products: + resolution = resolve_subscription_offerings( + selector, + self._offerings, + product_offering_ids=offering_ids, + destination=destination, + at=now, + ) + if not resolution.offerings: + unresolved.append(package_id) + if len(unresolved) == len(products): + entries.append( + _Exclusion("offering_unsupported", now, media_buy_id=buy.media_buy_id) + ) + elif unresolved: + entries.append( + _Exclusion( + "offering_unsupported", + now, + media_buy_id=buy.media_buy_id, + package_ids=tuple(sorted(unresolved)), + ) + ) + return entries + + def _sync_exclusions( + self, + principal: _PrincipalState, + subscription_id: str, + account_id: str, + wanted: Sequence[_Exclusion], + state: Mapping[str, Any], + accounts: Mapping[str, _Account], + buys: Sequence[SubscriptionBuy], + destinations: Mapping[str, Any], + now: datetime, + ) -> None: + entries = principal.exclusions.setdefault((subscription_id, account_id), []) + open_keys = {entry.key: entry for entry in entries if entry.until is None} + wanted_keys = {entry.key for entry in wanted} + account = accounts.get(account_id) + cutover: datetime | None = None + for key, entry in open_keys.items(): + if key not in wanted_keys: + cutover = cutover or self._cutover(state, account, buys, destinations, now) + entry.until = cutover + for entry in wanted: + if entry.key in open_keys: + continue + if entry.effective_at is not None and entry.effective_at <= now: + entry.from_ = self._cutover(state, account, buys, destinations, entry.effective_at) + else: + cutover = cutover or self._cutover(state, account, buys, destinations, now) + entry.from_ = cutover + entries.append(entry) + + def _schedules( + self, + state: Mapping[str, Any], + account: _Account | None, + buys: Sequence[SubscriptionBuy], + destinations: Mapping[str, Any], + now: datetime, + ) -> list[tuple[str, PeriodCalendar]]: + """``(offering_id, calendar)`` for each offering the account resolves to.""" + if account is None: + return [] + selector = SubscriptionSelector.from_subscription(state["configuration"]) + destination = _destination_for(selector, destinations) + products = [ids for buy in buys for _package, ids in buy.product_offering_ids()] + if not products: + products = [tuple(item.offering_id for item in self._offerings)] + found: dict[str, SubscriptionOffering] = {} + for offering_ids in products: + # Per product, as the coverage rule resolves it. + for item in resolve_subscription_offerings( + selector, + self._offerings, + product_offering_ids=offering_ids, + destination=destination, + at=now, + ).offerings: + found[item.offering_id] = item + return [ + (item.offering_id, item.calendar(account.timezone, selector.period_timezone)) + for item in sorted(found.values(), key=lambda value: value.offering_id.encode()) + ] + + def _cutover( + self, + state: Mapping[str, Any], + account: _Account | None, + buys: Sequence[SubscriptionBuy], + destinations: Mapping[str, Any], + now: datetime, + ) -> datetime: + """The cutover instant for an account-level exception changing at ``now``. + + The earliest E among the resolved schedules: no schedule has a boundary + in ``[now, E)``, so nothing is owed twice and nothing is lost, whereas + the latest E would leave a finer schedule owing periods after the + change. + """ + schedules = self._schedules(state, account, buys, destinations, now) + if not schedules: + return now + return min(calendar.ceil(now) for _offering_id, calendar in schedules) + + def _accounts_with_exceptions( + self, principal: _PrincipalState, subscription_id: str, now: datetime + ) -> list[dict[str, Any]]: + result = [] + for (sub_id, account_id), entries in sorted(principal.exclusions.items()): + if sub_id != subscription_id: + continue + account = principal.accounts.get(account_id) + if account is None or not account.authorized: + continue + reasons = sorted({entry.reason for entry in entries if entry.until is None}) + if reasons: + result.append({"account_id": account_id, "reasons": reasons}) + return result + + def _accounts_unauthorized( + self, + principal: _PrincipalState, + state: Mapping[str, Any], + accounts: Mapping[str, _Account], + buys: Mapping[str, Sequence[SubscriptionBuy]], + destinations: Mapping[str, Any], + now: datetime, + ) -> list[dict[str, Any]]: + subscription_id = str(state["subscription_id"]) + result = [] + for account_id in sorted(principal.covered.get(subscription_id, ())): + account = accounts.get(account_id) + if account is None or account.authorized or not account.grants: + continue + lost = account.grants[-1].unauthorized_at + assert lost is not None + if now - lost >= self._retention: + continue + schedules = self._schedules(state, account, (), destinations, lost) + owed = self._owed( + principal, + state, + account_id, + account, + buys.get(account_id, ()), + destinations, + now, + ) + entry: dict[str, Any] = { + "account_id": account_id, + "unauthorized_at": iso_z(lost), + "withheld_period_count": sum(1 for item in owed if item.withheld), + # Owed periods that ended by the loss, for a delivery hook. + "_ended": [ + item + for item in owed + if datetime.fromisoformat(item.period_end.replace("Z", "+00:00")) <= lost + ], + } + if schedules: + entry["last_owed_period_end"] = iso_z( + max(calendar.floor(lost) for _offering_id, calendar in schedules) + ) + result.append(entry) + return result + + # -- reads ----------------------------------------------------------------- + + async def subscription_coverage( + self, + identity: PrincipalIdentity, + account_id: str, + *, + now: datetime | None = None, + adcp_version: str | None = None, + ) -> list[ReportingSubscriptionCoverage]: + """``subscription_coverage[]`` for one account, while the caller is authorized. + + One entry per caller subscription whose scope includes the account, + inactive ones included while retained. An unauthorized account is + unknown here, as for any unauthorized caller. ``adcp_version`` is the + release the response is served at: below 3.3.0-beta.2, the adcp#8585 + fields are dropped and per-offering gaps and cutovers become + account-wide (see :mod:`adcp.reporting.subscriptions.wire`). + """ + instant = self._now(now) + states = [_dump(item) for item in await self._principals.reporting_subscriptions(identity)] + destinations = await self._destinations(identity) + buys = await self._host.account_buys(identity, account_id) + async with self._store.lock: + principal = self._store.principal(identity.subject) + account = principal.accounts.get(account_id) + if account is None or not account.authorized: + return [] + result = [] + for state in states: + configuration = state["configuration"] + if not _scope_includes(configuration, account_id, account): + continue + entries = principal.exclusions.get((str(state["subscription_id"]), account_id), []) + account_level = [e for e in entries if e.until is None and e.media_buy_id is None] + status = "active" + if any(e.reason == "action_required" for e in account_level): + status = "action_required" + elif account_level: + status = "excluded" + payload: dict[str, Any] = { + "subscription_id": state["subscription_id"], + "subscription_version": state["subscription_version"], + "status": status, + "excluded": [ + entry.wire() + for entry in entries + if entry.until is None or instant - entry.until <= self._retention + ], + } + gaps = self._gaps(state, account, buys, destinations, instant) + if gaps: + payload["gaps"] = gaps + cutovers = self._timezone_cutovers(state, account, buys, destinations, instant) + if cutovers: + payload["timezone_cutovers"] = cutovers + result.append( + ReportingSubscriptionCoverage.model_validate( + project_subscription_coverage(payload, adcp_version) + ) + ) + return result + + def _gaps( + self, + state: Mapping[str, Any], + account: _Account, + buys: Sequence[SubscriptionBuy], + destinations: Mapping[str, Any], + now: datetime, + ) -> list[dict[str, Any]]: + gaps = [] + for earlier, later in zip(account.grants, account.grants[1:]): + lost = earlier.unauthorized_at + assert lost is not None + regranted = later.authorized_at + expired = regranted - lost >= self._retention + if self._support.backfill_on_regrant and not expired: + continue + if expired: + continue # a fresh coverage start claims no gap + schedules = self._schedules(state, account, buys, destinations, lost) + entries = [ + (offering_id, calendar.floor(lost), calendar.ceil(regranted)) + for offering_id, calendar in schedules + ] + if not entries or now - max(until for _o, _f, until in entries) > self._retention: + continue + gaps.extend( + _per_offering( + entries, + lambda start, until: { + "from": iso_z(start), + "until": iso_z(until), + "reason": "unauthorized", + }, + ) + ) + return gaps + + def _timezone_cutovers( + self, + state: Mapping[str, Any], + account: _Account, + buys: Sequence[SubscriptionBuy], + destinations: Mapping[str, Any], + now: datetime, + ) -> list[dict[str, Any]]: + cutovers = [] + for change in account.timezone_changes: + # Each change carries the zone it replaced, which the wire reports. + zone = change.previous_timezone + selector = SubscriptionSelector.from_subscription(state["configuration"]) + resolution = resolve_subscription_offerings( + selector, + self._offerings, + product_offering_ids=_buy_offering_ids(buys) + or [item.offering_id for item in self._offerings], + destination=_destination_for(selector, destinations), + at=change.changed_at, + ) + effective = [ + ( + item.offering_id, + item.calendar(zone, selector.period_timezone).ceil(change.changed_at), + ) + for item in resolution.offerings + if item.alignment == "account_timezone" + ] + if effective and now - change.changed_at <= self._retention: + previous, timezone = zone, change.timezone + + def cutover(at: datetime, _same: datetime) -> dict[str, Any]: + return { + "effective_at": iso_z(at), + "timezone": timezone, + "previous_timezone": previous, + } + + cutovers.extend( + _per_offering([(offering_id, at, at) for offering_id, at in effective], cutover) + ) + return cutovers + + async def owed_periods( + self, + identity: PrincipalIdentity, + subscription_id: str, + account_id: str, + *, + now: datetime | None = None, + ) -> list[ExpectedSubscriptionPeriod]: + """Every period this subscription owes the account: the seller's obligations. + + The same :func:`expected_subscription_periods` a buyer runs, over the + engine's records: every version's configuration, the grant history, + timezone changes and every recorded exclusion. + """ + instant = self._now(now) + state = next( + ( + _dump(item) + for item in await self._principals.reporting_subscriptions(identity) + if item.subscription_id == subscription_id + ), + None, + ) + if state is None: + raise KeyError(subscription_id) + destinations = await self._destinations(identity) + buys = await self._host.account_buys(identity, account_id) + async with self._store.lock: + principal = self._store.principal(identity.subject) + account = principal.accounts.get(account_id) + if account is None: + return [] + return self._owed(principal, state, account_id, account, buys, destinations, instant) + + def _owed( + self, + principal: _PrincipalState, + state: Mapping[str, Any], + account_id: str, + account: _Account, + buys: Sequence[SubscriptionBuy], + destinations: Mapping[str, Any], + now: datetime, + ) -> list[ExpectedSubscriptionPeriod]: + subscription_id = str(state["subscription_id"]) + versions = { + version: configuration + for (sub_id, version), configuration in principal.versions.items() + if sub_id == subscription_id + } + exclusions = [ + SubscriptionException(e.reason, e.from_, e.until, e.media_buy_id) + for e in principal.exclusions.get((subscription_id, account_id), []) + if not e.package_ids + ] + # The seller never guesses a prior version's configuration, and each + # version resolves against the destination it named when accepted. + windows: tuple[SubscriptionVersionWindow, ...] = tuple( + replace( + window, + destination=principal.destinations.get( + (subscription_id, window.subscription_version) + ), + ) + for window in subscription_version_windows( + state, version_configurations=versions, require_prior_configurations=True + ) + ) + return expected_subscription_periods( + subscription=windows, + account_id=account_id, + buys=buys, + offerings=self._offerings, + account_timezone=account.timezone, + now=now, + authorizations=list(account.grants), + exceptions=exclusions, + backfill_on_regrant=self._support.backfill_on_regrant, + status_retention_days=self._retention_days, + destinations=destinations, + account_operator=account.operator, + timezone_changes=list(account.timezone_changes), + ) + + async def receipt_roster(self, identity: PrincipalIdentity, account_id: str) -> list[str]: + """Active ``consumer_receipt`` subscriptions covering the account. + + A consumer owes receipts for an account while one of its active + ``consumer_receipt`` subscriptions covers it without an open + account-level exception. + """ + states = [_dump(item) for item in await self._principals.reporting_subscriptions(identity)] + async with self._store.lock: + principal = self._store.principal(identity.subject) + account = principal.accounts.get(account_id) + if account is None or not account.authorized: + return [] + result = [] + for state in states: + configuration = state["configuration"] + if ( + not _active(state) + or configuration["reconciliation_mode"] != "consumer_receipt" + or not _scope_includes(configuration, account_id, account) + ): + continue + entries = principal.exclusions.get((str(state["subscription_id"]), account_id), []) + if any(e.until is None and e.media_buy_id is None for e in entries): + continue + result.append(str(state["subscription_id"])) + return sorted(result) + + +# -- helpers ------------------------------------------------------------------- + + +def _buy_offering_ids(buys: Sequence[SubscriptionBuy]) -> list[str]: + ids: set[str] = set() + for buy in buys: + for _package, offering_ids in buy.product_offering_ids(): + ids.update(offering_ids) + return sorted(ids) + + +def _per_offering( + entries: Sequence[tuple[str, datetime, datetime]], + build: Callable[[datetime, datetime], dict[str, Any]], +) -> list[dict[str, Any]]: + """One entry for every offering when their boundaries agree, else one each. + + ``gaps[]`` and ``timezone_cutovers[]`` entries without ``offering_id`` + apply to every offering on the account (adcp#8585). + """ + instants = {(first, second) for _offering_id, first, second in entries} + if len(instants) == 1: + first, second = next(iter(instants)) + return [build(first, second)] + result = [] + for offering_id, first, second in sorted(entries, key=lambda item: item[0].encode()): + entry = build(first, second) + entry["offering_id"] = offering_id + result.append(entry) + return result + + +def _operator_selected(configuration: Mapping[str, Any], account: _Account) -> bool: + domains = configuration.get("operator_domains") + return not domains or account.operator in set(domains) + + +def _scope_selects(configuration: Mapping[str, Any], account_id: str, account: _Account) -> bool: + try: + return subscription_covers_account( + configuration, account_id=account_id, account_operator=account.operator + ) + except ValueError: + return False # operator_domains with an account the host gave no operator + + +def _scope_includes(configuration: Mapping[str, Any], account_id: str, account: _Account) -> bool: + """Whether the scope names the account, counting buyer exclusions as included.""" + excluded = set(configuration.get("exclude_account_ids") or ()) + if account_id in excluded: + return _operator_selected(configuration, account) + return _scope_selects(configuration, account_id, account) + + +def _override( + configuration: Mapping[str, Any], + scope: Mapping[str, Any], + feed_purpose: str, + account_id: str, + account: _Account, + states: Sequence[Mapping[str, Any]], + configs: Sequence[AccountDeliveryConfig], +) -> tuple[str, str, datetime | None] | None: + """The caller's own registration that takes this account and purpose, if any.""" + for config in sorted(configs, key=lambda item: item.delivery_config_id.encode("utf-8")): + if config.feed_purpose == feed_purpose: + return ("delivery_config_id", config.delivery_config_id, config.active_from) + if "all_authorized_accounts" in scope: + for other in states: + other_config = other["configuration"] + if ( + _active(other) + and "account_ids" in other_config["scope"] + and other_config["selector"]["feed_purpose"] == feed_purpose + and _scope_selects(other_config, account_id, account) + ): + return ("subscription_id", str(other["subscription_id"]), None) + return None + + +def _destination_for( + selector: SubscriptionSelector, destinations: Mapping[str, Any] +) -> SubscriptionDestination | None: + if selector.destination_id is None: + return None + target = destinations.get(selector.destination_id) + return None if target is None else SubscriptionDestination.coerce(target) + + +__all__ = [ + "AccountDeliveryConfig", + "AuthorizedAccount", + "AuthorizedAccountPage", + "InMemoryReportingSubscriptionStore", + "ReportingSubscriptionDeliveryHost", + "ReportingSubscriptionEngine", + "ReportingSubscriptionHost", + "ReportingSubscriptionPrecedenceHost", +] diff --git a/src/adcp/reporting/subscriptions/registry.py b/src/adcp/reporting/subscriptions/registry.py new file mode 100644 index 000000000..7da09cdd8 --- /dev/null +++ b/src/adcp/reporting/subscriptions/registry.py @@ -0,0 +1,488 @@ +"""Accept, validate and version the ``sync_principal`` ``reporting_subscriptions`` section. + +The section is complete desired state, like the principal's other sections. +:func:`apply_reporting_subscriptions` turns the caller's desired list and the +previously stored states into the new ``core/reporting-subscription-state.json`` +states, or raises :class:`ReportingSubscriptionError` so the whole request +fails. It enforces what the generated model cannot express (the schema's +``allOf`` conditionals and ``x-adcp-validation`` rules): + +* ``scope`` is exactly one of ``all_authorized_accounts: true`` or + ``account_ids``; ``exclude_account_ids`` only with all-accounts scope; +* ``period_timezone`` only when ``account_ids`` names exactly one account, and + it must be a recognized IANA zone; +* billing requires ``official`` finality and ``consumer_receipt``, and + ``consumer_receipt`` requires a destination; +* ``destination_id`` names a destination that exists after the same request + applies; ``path_template`` uses only the four placeholders, contains + ``{account_id}`` when the scope can cover several accounts or another + active subscription shares the destination (and is then required for + ``file_transfer``), and shared templates differ; +* ``subscription_id`` values are unique and never equal a ``delivery_config_id`` + the caller uses through ``sync_accounts``; +* two active subscriptions at the same scope level may not both apply to one + account and ``feed_purpose``. + +Versioning: a ``subscription_version`` is immutable. Resending it with +different semantic content is rejected; ``active`` and +``replace_account_configs`` change in place. A higher version records its own +``accepted_at`` and moves the old one to ``prior_versions`` with +``superseded_at``. Omitting a subscription, or ``active: false``, deactivates +it at ``deactivated_at``; it stays readable as ``inactive``. +""" + +from __future__ import annotations + +import re +from collections.abc import Collection, Mapping, Sequence +from datetime import datetime +from typing import Any +from urllib.parse import quote + +from adcp.reporting.subscriptions._clock import iso_z, require_zone, utc_instant +from adcp.types.domains.core.reporting_subscription import ReportingSubscription +from adcp.types.domains.core.reporting_subscription_state import ReportingSubscriptionState + +_PLACEHOLDER = re.compile(r"\{([^{}]*)\}") +_ALLOWED_PLACEHOLDERS = frozenset({"account_id", "subscription_id", "offering_id", "period_start"}) +#: Fields that may change without a new ``subscription_version``. +_IN_PLACE_FIELDS = frozenset({"active", "replace_account_configs", "subscription_version"}) + + +class ReportingSubscriptionError(ValueError): + """The section is invalid; the whole ``sync_principal`` request fails.""" + + def __init__(self, message: str, *, code: str = "INVALID_REQUEST") -> None: + super().__init__(message) + self.code = code + + +def _dump(value: Any) -> dict[str, Any]: + if hasattr(value, "model_dump"): + dumped = value.model_dump(mode="json", exclude_none=True) + assert isinstance(dumped, dict) + return dumped + return dict(value) + + +def _semantic(configuration: Mapping[str, Any]) -> dict[str, Any]: + """The semantic content of a version, canonicalized. + + ``uniqueItems`` arrays compare as sets and an explicit default ``match`` + equals an omitted one, so a resend that only reorders or spells out a + default is the same version. + """ + payload = {key: value for key, value in configuration.items() if key not in _IN_PLACE_FIELDS} + scope = dict(payload.get("scope") or {}) + if "account_ids" in scope: + scope["account_ids"] = sorted(scope["account_ids"]) + payload["scope"] = scope + for name in ("exclude_account_ids", "operator_domains"): + if name in payload: + payload[name] = sorted(payload[name]) + selector = dict(payload.get("selector") or {}) + if "dimensions" in selector: + selector["dimensions"] = sorted(selector["dimensions"]) + selector.setdefault("match", "exact" if "dimensions" in selector else "finest") + payload["selector"] = selector + return payload + + +def _destinations(destinations: Sequence[Any]) -> dict[str, dict[str, Any]]: + result: dict[str, dict[str, Any]] = {} + for item in destinations: + payload = _dump(item) + result[str(payload["destination_id"])] = payload + return result + + +def _multi_account(configuration: Mapping[str, Any]) -> bool: + scope = configuration.get("scope") or {} + return scope.get("all_authorized_accounts") is True or len(scope.get("account_ids") or ()) > 1 + + +def _validate_one( + configuration: dict[str, Any], + index: int, + *, + destinations: Mapping[str, Mapping[str, Any]], + delivery_config_ids: Collection[str], +) -> None: + where = f"reporting_subscriptions[{index}]" + subscription_id = configuration["subscription_id"] + if subscription_id in delivery_config_ids: + raise ReportingSubscriptionError( + f"{where}.subscription_id equals a delivery_config_id this caller uses; " + "subscription_id and delivery_config_id share one namespace" + ) + scope = configuration.get("scope") or {} + has_all = "all_authorized_accounts" in scope + has_ids = "account_ids" in scope + if has_all == has_ids: + raise ReportingSubscriptionError( + f"{where}.scope must be exactly one of all_authorized_accounts or account_ids" + ) + if has_all and scope["all_authorized_accounts"] is not True: + raise ReportingSubscriptionError(f"{where}.scope.all_authorized_accounts must be true") + if configuration.get("exclude_account_ids") and not has_all: + raise ReportingSubscriptionError( + f"{where}.exclude_account_ids is valid only with all_authorized_accounts scope" + ) + if configuration.get("replace_account_configs") is True: + # Not applied yet: accepting it would leave the caller's account-level + # configurations in force while it believes they were cut over. + raise ReportingSubscriptionError( + f"{where}.replace_account_configs is not supported by this seller", + code="UNSUPPORTED_FEATURE", + ) + selector = configuration["selector"] + period_timezone = selector.get("period_timezone") + if period_timezone is not None: + if not has_ids or len(scope["account_ids"]) != 1: + raise ReportingSubscriptionError( + f"{where}.selector.period_timezone requires account_ids naming exactly one account" + ) + try: + require_zone(period_timezone, field="period_timezone") + except ValueError as error: + raise ReportingSubscriptionError(f"{where}.selector.{error}") from None + mode = configuration["reconciliation_mode"] + if selector["feed_purpose"] == "billing" and ( + selector["required_finality"] != "official" or mode != "consumer_receipt" + ): + raise ReportingSubscriptionError( + f"{where}: billing subscriptions require official finality and consumer_receipt" + ) + destination = configuration.get("destination") + if mode == "consumer_receipt" and destination is None: + raise ReportingSubscriptionError(f"{where}: consumer_receipt requires a destination") + if destination is None: + return + target = destinations.get(destination["destination_id"]) + if target is None: + raise ReportingSubscriptionError( + f"{where}.destination.destination_id {destination['destination_id']!r} is not in " + "reporting_destinations after this request" + ) + template = destination.get("path_template") + if template is not None: + unknown = {name for name in _PLACEHOLDER.findall(template)} - _ALLOWED_PLACEHOLDERS + if unknown: + raise ReportingSubscriptionError( + f"{where}.destination.path_template uses unknown placeholders {sorted(unknown)}" + ) + if "{" in _PLACEHOLDER.sub("", template) or "}" in _PLACEHOLDER.sub("", template): + raise ReportingSubscriptionError( + f"{where}.destination.path_template has braces outside a placeholder" + ) + if _multi_account(configuration): + pattern = (target.get("configuration") or target).get("pattern") + if template is None and pattern == "file_transfer": + raise ReportingSubscriptionError( + f"{where}.destination.path_template is required for a multi-account " + "file_transfer subscription" + ) + if template is not None and "{account_id}" not in template: + raise ReportingSubscriptionError( + f"{where}.destination.path_template must contain {{account_id}} when the " + "scope can cover more than one account" + ) + + +def _validate_set(active: Sequence[dict[str, Any]], destinations: Mapping[str, Any]) -> None: + by_destination: dict[str, list[dict[str, Any]]] = {} + for configuration in active: + destination = configuration.get("destination") + if destination is not None: + by_destination.setdefault(destination["destination_id"], []).append(configuration) + for destination_id, sharing in by_destination.items(): + if len(sharing) < 2: + continue + templates = [(item.get("destination") or {}).get("path_template") for item in sharing] + if any(template is None or "{account_id}" not in template for template in templates): + raise ReportingSubscriptionError( + f"active subscriptions sharing destination {destination_id!r} must each have a " + "path_template containing {account_id}" + ) + if len(set(templates)) != len(templates): + raise ReportingSubscriptionError( + f"active subscriptions sharing destination {destination_id!r} must use " + "different path_template values" + ) + for first_index, first in enumerate(active): + for second in active[first_index + 1 :]: + if first["selector"]["feed_purpose"] != second["selector"]["feed_purpose"]: + continue + if _same_level_overlap(first, second): + raise ReportingSubscriptionError( + f"active subscriptions {first['subscription_id']!r} and " + f"{second['subscription_id']!r} would both apply to one account and " + "feed_purpose at the same scope level" + ) + + +def _same_level_overlap(first: Mapping[str, Any], second: Mapping[str, Any]) -> bool: + first_scope, second_scope = first["scope"], second["scope"] + if "account_ids" in first_scope and "account_ids" in second_scope: + shared = set(first_scope["account_ids"]) & set(second_scope["account_ids"]) + if not shared: + return False + return _operator_domains_overlap(first, second) + if "all_authorized_accounts" in first_scope and "all_authorized_accounts" in second_scope: + # exclude_account_ids cannot make two all-accounts scopes disjoint for + # accounts granted later; disjoint operator_domains can. + return _operator_domains_overlap(first, second) + return False + + +def _operator_domains_overlap(first: Mapping[str, Any], second: Mapping[str, Any]) -> bool: + first_domains = first.get("operator_domains") + second_domains = second.get("operator_domains") + if not first_domains or not second_domains: + return True + return bool(set(first_domains) & set(second_domains)) + + +def apply_reporting_subscriptions( + previous: Sequence[ReportingSubscriptionState | Mapping[str, Any]], + desired: Sequence[ReportingSubscription | Mapping[str, Any]], + *, + destinations: Sequence[Any], + now: datetime, + delivery_config_ids: Collection[str] = (), +) -> list[ReportingSubscriptionState]: + """Return the new subscription states for a ``sync_principal`` request. + + ``destinations`` are the caller's ``reporting_destinations`` states after + this request applies. ``delivery_config_ids`` are every + ``delivery_config_id`` the caller uses through ``sync_accounts``. + Engine-derived fields (``accounts_with_exceptions``, + ``accounts_unauthorized``, ``issues``) carry over from ``previous``. + """ + now = utc_instant(now, field="now") + stamp = iso_z(now) + by_id = {str(_dump(item)["subscription_id"]): _dump(item) for item in previous} + targets = _destinations(destinations) + configurations = [_dump(item) for item in desired] + seen: set[str] = set() + for index, configuration in enumerate(configurations): + subscription_id = configuration["subscription_id"] + if subscription_id in seen: + raise ReportingSubscriptionError( + f"reporting_subscriptions[{index}].subscription_id duplicates an earlier entry" + ) + seen.add(subscription_id) + _validate_one( + configuration, + index, + destinations=targets, + delivery_config_ids=delivery_config_ids, + ) + _validate_set([item for item in configurations if item.get("active", True)], targets) + + states: list[ReportingSubscriptionState] = [] + for configuration in configurations: + old = by_id.pop(configuration["subscription_id"], None) + states.append(_next_state(old, configuration, stamp=stamp, destinations=targets)) + for old in by_id.values(): + # Omitted: deactivated by cutover now, retained as inactive for history. + states.append(_deactivated(old, stamp=stamp)) + return [ReportingSubscriptionState.model_validate(item) for item in _sorted(states)] + + +def _sorted(states: Sequence[ReportingSubscriptionState]) -> list[dict[str, Any]]: + return sorted((_dump(item) for item in states), key=lambda item: item["subscription_id"]) + + +def _deactivated(old: dict[str, Any], *, stamp: str) -> ReportingSubscriptionState: + payload = dict(old) + if payload.get("state") != "inactive": + configuration = dict(payload["configuration"]) + configuration["active"] = False + payload.update(state="inactive", deactivated_at=stamp, configuration=configuration) + payload.pop("issues", None) + payload["inactive_intervals"] = _intervals( + [*(old.get("inactive_intervals") or ()), {"deactivated_at": stamp}] + ) + return ReportingSubscriptionState.model_validate(payload) + + +#: ``inactive_intervals`` holds at most this many entries (the schema's maxItems). +_MAX_INACTIVE_INTERVALS = 64 + + +def _intervals(intervals: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: + """Every interval, or an error: dropping one would make its periods owed again.""" + if len(intervals) > _MAX_INACTIVE_INTERVALS: + raise ReportingSubscriptionError( + f"the subscription has reached {_MAX_INACTIVE_INTERVALS} inactive intervals, the " + "readback's limit; register a new subscription_id instead of deactivating again" + ) + return [dict(item) for item in intervals] + + +def _next_state( + old: dict[str, Any] | None, + configuration: dict[str, Any], + *, + stamp: str, + destinations: Mapping[str, Mapping[str, Any]], +) -> ReportingSubscriptionState: + version = int(configuration["subscription_version"]) + active = bool(configuration.get("active", True)) + payload: dict[str, Any] = { + "subscription_id": configuration["subscription_id"], + "subscription_version": version, + "configuration": configuration, + } + if old is None: + payload["accepted_at"] = stamp + else: + old_version = int(old["subscription_version"]) + if version < old_version: + raise ReportingSubscriptionError( + f"subscription {configuration['subscription_id']!r} version {version} is older " + f"than the accepted version {old_version}" + ) + if version == old_version: + if _semantic(configuration) != _semantic(old["configuration"]): + raise ReportingSubscriptionError( + f"subscription {configuration['subscription_id']!r} reuses version " + f"{version} with different semantic content; increment subscription_version" + ) + payload["accepted_at"] = old["accepted_at"] + if old.get("prior_versions"): + payload["prior_versions"] = old["prior_versions"] + else: + payload["accepted_at"] = stamp + payload["prior_versions"] = [ + { + "subscription_version": old_version, + "accepted_at": old["accepted_at"], + "superseded_at": stamp, + # adcp#8588: sellers MUST include the version's configuration. + "configuration": old["configuration"], + }, + *(old.get("prior_versions") or ()), + ] + for carried in ("accounts_with_exceptions", "accounts_unauthorized"): + if old.get(carried): + payload[carried] = old[carried] + # inactive_intervals (adcp#8585) let a buyer recompute a reactivation in + # place, which the readback otherwise loses with deactivated_at. + intervals = [dict(item) for item in (old or {}).get("inactive_intervals") or ()] + was_inactive = old is not None and old.get("state") == "inactive" + if active: + payload["state"] = "active" + if was_inactive and intervals and "reactivated_at" not in intervals[-1]: + intervals[-1]["reactivated_at"] = stamp + else: + payload["state"] = "inactive" + if not was_inactive or not intervals or "reactivated_at" in intervals[-1]: + start = old["deactivated_at"] if was_inactive and old is not None else stamp + intervals.append({"deactivated_at": start}) + payload["deactivated_at"] = intervals[-1]["deactivated_at"] + if intervals: + payload["inactive_intervals"] = _intervals(intervals) + destination = configuration.get("destination") + if destination is not None: + target = destinations[destination["destination_id"]] + payload["destination_ref"] = target["destination_ref"] + if active: + issue = _destination_issue(target) + if issue is not None: + payload["state"] = "action_required" + payload["issues"] = [issue] + return ReportingSubscriptionState.model_validate(payload) + + +def _destination_issue(target: Mapping[str, Any]) -> dict[str, Any] | None: + """A destination the caller must act on blocks delivery everywhere.""" + state = str(target.get("state")) + if state in {"action_required", "inactive", "rejected"}: + return { + "code": "DESTINATION_NOT_READY", + "message": f"destination {target['destination_id']!r} is {state}", + "field": "destination.destination_id", + } + return None + + +def render_path_template( + template: str, + *, + account_id: str, + subscription_id: str, + offering_id: str, + period_start: datetime, +) -> str: + """Expand a subscription ``path_template`` for one delivery. + + Each substituted value fills exactly one path segment: a value containing + ``/`` (or ``\\``, for backslash-separated targets) is percent-encoded + (RFC 3986), and a value equal to ``.`` or ``..`` is encoded too, so no + value can add or escape a segment. + ``period_start`` is written as ``YYYY-MM-DDTHH:MM:SSZ``. + """ + values = { + "account_id": account_id, + "subscription_id": subscription_id, + "offering_id": offering_id, + "period_start": iso_z(period_start), + } + + def segment(match: re.Match[str]) -> str: + name = match.group(1) + if name not in values: + raise ValueError(f"unknown path_template placeholder {{{name}}}") + value = values[name] + if value in {".", ".."}: + return value.replace(".", "%2E") + if "/" in value or "\\" in value: + # "/" per the spec; "\\" too, so a backslash-separated target + # (SMB, UNC) cannot gain or escape a segment either. + return quote(value, safe="") + return value + + return _PLACEHOLDER.sub(segment, template) + + +def refresh_destination_states( + states: Sequence[ReportingSubscriptionState | Mapping[str, Any]], + destinations: Sequence[Any], +) -> list[ReportingSubscriptionState]: + """Recompute ``state``, ``issues`` and ``destination_ref`` after a destination changes.""" + targets = _destinations(destinations) + refreshed: list[ReportingSubscriptionState] = [] + for item in states: + payload = _dump(item) + destination = payload["configuration"].get("destination") + if payload.get("state") != "inactive" and destination is not None: + target = targets.get(destination["destination_id"]) + issue = ( + { + "code": "DESTINATION_NOT_READY", + "message": f"destination {destination['destination_id']!r} was removed", + "field": "destination.destination_id", + } + if target is None + else _destination_issue(target) + ) + if target is not None: + payload["destination_ref"] = target["destination_ref"] + if issue is None: + payload["state"] = "active" + payload.pop("issues", None) + else: + payload["state"] = "action_required" + payload["issues"] = [issue] + refreshed.append(ReportingSubscriptionState.model_validate(payload)) + return refreshed + + +__all__ = [ + "ReportingSubscriptionError", + "apply_reporting_subscriptions", + "refresh_destination_states", + "render_path_template", +] diff --git a/src/adcp/server/__init__.py b/src/adcp/server/__init__.py index b9ef2b5d6..9c95aa966 100644 --- a/src/adcp/server/__init__.py +++ b/src/adcp/server/__init__.py @@ -146,6 +146,7 @@ async def get_products(params, context=None): PrincipalRecord, PrincipalRecordStore, PrincipalService, + ReportingSubscriptionOptions, ) from adcp.server.proposal import ProposalBuilder, ProposalNotSupported from adcp.server.responses import ( @@ -244,6 +245,7 @@ async def get_products(params, context=None): "PrincipalRecord", "PrincipalRecordStore", "PrincipalService", + "ReportingSubscriptionOptions", # Proposal helpers "ProposalBuilder", "ProposalNotSupported", diff --git a/src/adcp/server/principal.py b/src/adcp/server/principal.py index 4980c80e9..03d69a70e 100644 --- a/src/adcp/server/principal.py +++ b/src/adcp/server/principal.py @@ -12,10 +12,10 @@ import json import re import unicodedata -from collections.abc import Awaitable, Callable, Mapping, Sequence +from collections.abc import Awaitable, Callable, Collection, Mapping, Sequence from dataclasses import dataclass, field from datetime import datetime, timezone -from typing import TYPE_CHECKING, Any, Protocol +from typing import TYPE_CHECKING, Any, Protocol, TypeVar from urllib.parse import urlsplit, urlunsplit from uuid import uuid4 @@ -38,10 +38,13 @@ SyncPrincipalRequest, SyncPrincipalResponse, ) +from adcp.types.domains.core.reporting_subscription_state import ReportingSubscriptionState if TYPE_CHECKING: from adcp.server.base import ToolContext +_ResponseT = TypeVar("_ResponseT", GetPrincipalResponse, SyncPrincipalResponse) + # get_principal relationship is added in AdCP 3.3 (DR-0024): result arms served # below 3.3 are closed, so the field is emitted only at 3.3 or later. _RELATIONSHIP_MIN_RELEASE = (3, 3) @@ -89,6 +92,8 @@ class PrincipalChange: changed_at: datetime reason: str destination_id: str | None = None + #: Set for ``reporting_subscription_changed``: the subscription that changed. + subscription_id: str | None = None PrincipalChangeEmitter = Callable[[PrincipalChange], Awaitable[None]] @@ -99,6 +104,28 @@ class PrincipalChange: NotificationProofHook = Callable[ [PrincipalIdentity, AgentNotificationConfig], Awaitable[AgentNotificationConfigState] ] +DeliveryConfigIdsHook = Callable[[PrincipalIdentity], Awaitable[Collection[str]]] +SubscriptionVersionsHook = Callable[ + [PrincipalIdentity, Sequence[ReportingSubscriptionState]], Awaitable[None] +] + + +@dataclass(frozen=True) +class ReportingSubscriptionOptions: + """Opt-in acceptance of the 3.3 ``reporting_subscriptions`` section. + + Without these options ``PrincipalService`` keeps its 3.2 behavior and + ignores the section. ``delivery_config_ids`` returns every + ``delivery_config_id`` the caller uses through ``sync_accounts`` so a + ``subscription_id`` never collides with one. ``record_versions`` receives + every committed subscription state, so the reporting runtime keeps each + version's configuration for as long as its obligations are retained. + ``ReportingSubscriptionEngine.principal_options()`` builds these. + """ + + delivery_config_ids: DeliveryConfigIdsHook | None = None + record_versions: SubscriptionVersionsHook | None = None + clock: Callable[[], datetime] | None = None @dataclass @@ -178,6 +205,37 @@ def _copy_record(record: PrincipalRecord) -> PrincipalRecord: ) +def _gate_subscription_fields(response: _ResponseT, context: ToolContext | None) -> _ResponseT: + """Drop adcp#8585 readback fields for callers served below 3.3.0-beta.2. + + The server validates responses against the release it serves, and + 3.3.0-beta.1 closes the subscription state object. + """ + from adcp.reporting.subscriptions.wire import ( + project_subscription_state, + serves_settled_subscription_fields, + ) + + configuration = getattr(response.result, "configuration", None) + states = getattr(configuration, "reporting_subscriptions", None) + if not states or serves_settled_subscription_fields( + getattr(context, "resolved_adcp_version", None) + ): + return response + if not any( + getattr(item, "inactive_intervals", None) + or any(getattr(prior, "configuration", None) for prior in item.prior_versions or ()) + for item in states + ): + return response + envelope = response.model_dump(mode="json", exclude_none=True) + envelope["result"]["configuration"]["reporting_subscriptions"] = [ + project_subscription_state(item, None) + for item in envelope["result"]["configuration"]["reporting_subscriptions"] + ] + return type(response).model_validate(envelope) + + def _json(value: object) -> str: if isinstance(value, BaseModel): value = value.model_dump(mode="json", exclude_none=True) @@ -320,6 +378,12 @@ def _normalize_coordinate(value: str, *, field: str) -> str: return urlunsplit((scheme, netloc, path, "", "")) +def _destination_id(destination: Any) -> str: + if isinstance(destination, Mapping): + return str(destination["destination_id"]) + return str(destination.destination_id) + + def _normalize_destination(destination: AgentReportingDestination) -> AgentReportingDestination: payload = destination.model_dump(mode="json", exclude_none=True) if "location" in payload: @@ -352,8 +416,10 @@ def __init__( relationship_resolver: ( Callable[[PrincipalIdentity], Awaitable[Mapping[str, Any] | BaseModel | None]] | None ) = None, + reporting_subscriptions: ReportingSubscriptionOptions | None = None, ) -> None: self._store = store or InMemoryPrincipalRecordStore() + self._subscriptions = reporting_subscriptions self._destination_proof = destination_proof self._notification_proof = notification_proof self._emit_change = emit_change @@ -364,6 +430,13 @@ def __init__( "experimental_features": accepted_experimental_features, } + def set_reporting_subscriptions(self, options: ReportingSubscriptionOptions | None) -> None: + """Enable (or, with ``None``, disable) the 3.3 ``reporting_subscriptions`` section. + + Lets a reporting engine built from this service link back to it. + """ + self._subscriptions = options + def set_declaration_support( self, *, @@ -408,7 +481,9 @@ async def get_principal( response keeps the closed pre-3.3 result shape. Pass the handler's ``ToolContext`` so the served release is known. """ - response = await self._get_principal_state(identity, request) + response = _gate_subscription_fields( + await self._get_principal_state(identity, request), context + ) if ( self._relationship_resolver is None or response.result.kind == "failed" @@ -472,6 +547,21 @@ async def sync_principal( self, identity: PrincipalIdentity, request: SyncPrincipalRequest | Mapping[str, Any], + *, + context: ToolContext | None = None, + ) -> SyncPrincipalResponse: + """Apply the submitted sections atomically. + + Pass the handler's ``ToolContext`` so fields a newer release added + (such as reporting-subscription ``inactive_intervals``) reach only + callers served that release. + """ + return _gate_subscription_fields(await self._sync_principal(identity, request), context) + + async def _sync_principal( + self, + identity: PrincipalIdentity, + request: SyncPrincipalRequest | Mapping[str, Any], ) -> SyncPrincipalResponse: request = SyncPrincipalRequest.model_validate(request) if not request.configuration.model_fields_set: @@ -568,6 +658,10 @@ async def sync_principal( if await self._store.compare_and_swap( identity.subject, expected_store_revision, record ): + if self._subscriptions is not None and self._subscriptions.record_versions: + await self._subscriptions.record_versions( + identity, list(next_state.reporting_subscriptions or []) + ) return response # A different process committed between our read and write. Repeat # from the authoritative record so stale fences reject and an exact @@ -619,6 +713,13 @@ async def transition_destination( destinations[index] = AgentReportingDestinationState.model_validate(payload) state_payload = record.configuration.model_dump(mode="json", exclude_none=True) state_payload["reporting_destinations"] = destinations + subscriptions_before = list(record.configuration.reporting_subscriptions or []) + subscriptions_after = subscriptions_before + if self._subscriptions is not None and subscriptions_before: + from adcp.reporting.subscriptions.registry import refresh_destination_states + + subscriptions_after = refresh_destination_states(subscriptions_before, destinations) + state_payload["reporting_subscriptions"] = subscriptions_after record.configuration = PrincipalState.model_validate(state_payload) expected_store_revision = record.store_revision record.store_revision += 1 @@ -637,6 +738,21 @@ async def transition_destination( destination_id, ) ) + # Moving to action_required fires principal.changed, and so does + # the recovery, so a cached readback never stays action_required. + before = {item.subscription_id: str(item.state) for item in subscriptions_before} + for item in subscriptions_after: + if str(item.state) != before.get(item.subscription_id): + await self._emit_change( + PrincipalChange( + record.principal_id, + identity.subject, + datetime.now(timezone.utc), + "reporting_subscription_changed", + destination_id, + subscription_id=item.subscription_id, + ) + ) return destinations[index] async def refresh_declarations( @@ -791,6 +907,15 @@ async def _replace_sections( payload["declarations"] = self._negotiate_declarations( desired.declarations or AgentDeclarations() ) + if self._subscriptions is not None: + await self._replace_subscriptions(identity, previous, desired, payload) + elif "reporting_subscriptions" in fields: + # Unsupported sections are rejected, not ignored: a buyer told + # "applied" would believe it registered and receive nothing. + raise _PrincipalValidationError( + "UNSUPPORTED_FEATURE", + "reporting_subscriptions is not a supported section on this seller", + ) state = PrincipalState.model_validate(payload) active_notifications = [ config for config in state.notification_configs or [] if config.active is not False @@ -806,6 +931,151 @@ async def _replace_sections( ) return state + async def _replace_subscriptions( + self, + identity: PrincipalIdentity, + previous: PrincipalState | None, + desired: PrincipalConfiguration, + payload: dict[str, Any], + ) -> None: + """Apply ``reporting_subscriptions`` against the destinations after this request.""" + from adcp.reporting.subscriptions.registry import ( + ReportingSubscriptionError, + apply_reporting_subscriptions, + refresh_destination_states, + ) + + assert self._subscriptions is not None + destinations = payload.get("reporting_destinations") or [] + prior = list((previous.reporting_subscriptions if previous else None) or []) + try: + if "reporting_subscriptions" in desired.model_fields_set: + hook = self._subscriptions.delivery_config_ids + payload["reporting_subscriptions"] = apply_reporting_subscriptions( + prior, + desired.reporting_subscriptions or [], + destinations=destinations, + now=self._now(), + delivery_config_ids=await hook(identity) if hook else (), + ) + elif prior and "reporting_destinations" in desired.model_fields_set: + names = {_destination_id(item) for item in destinations} + for item in prior: + configuration = item.configuration + if ( + str(item.state) != "inactive" + and configuration.destination is not None + and configuration.destination.destination_id not in names + ): + raise ReportingSubscriptionError( + f"subscription {item.subscription_id!r} names destination " + f"{configuration.destination.destination_id!r}, which this " + "request removes" + ) + payload["reporting_subscriptions"] = refresh_destination_states(prior, destinations) + except ReportingSubscriptionError as error: + raise _PrincipalValidationError(error.code, str(error)) from None + + def _now(self) -> datetime: + clock = self._subscriptions.clock if self._subscriptions else None + return clock() if clock else datetime.now(timezone.utc) + + async def update_subscription_status( + self, + identity: PrincipalIdentity, + subscription_id: str, + *, + accounts_with_exceptions: Sequence[Mapping[str, Any]] | None = None, + accounts_unauthorized: Sequence[Mapping[str, Any]] | None = None, + changed_at: datetime | None = None, + ) -> bool: + """Record seller-derived subscription status without version churn. + + Like destination proof transitions, this does not advance the + caller-owned ``configuration_version``. A change emits + ``principal.changed`` with reason ``reporting_subscription_changed``, + dated ``changed_at`` (the instant the seller recorded the transition; + buyers date continuing re-grants from it). Returns whether anything + changed. + """ + while True: + record = await self._store.get(identity.subject) + if not record or not record.configuration: + raise KeyError(subscription_id) + states = list(record.configuration.reporting_subscriptions or []) + index = next( + (i for i, item in enumerate(states) if item.subscription_id == subscription_id), + None, + ) + if index is None: + raise KeyError(subscription_id) + payload = states[index].model_dump(mode="json", exclude_none=True) + for name, value in ( + ("accounts_with_exceptions", accounts_with_exceptions), + ("accounts_unauthorized", accounts_unauthorized), + ): + if value is None: + continue + if value: + payload[name] = [dict(item) for item in value] + else: + payload.pop(name, None) + updated = ReportingSubscriptionState.model_validate(payload) + if _json(updated) == _json(states[index]): + return False + states[index] = updated + state_payload = record.configuration.model_dump(mode="json", exclude_none=True) + state_payload["reporting_subscriptions"] = states + record.configuration = PrincipalState.model_validate(state_payload) + expected_store_revision = record.store_revision + record.store_revision += 1 + if await self._store.compare_and_swap( + identity.subject, expected_store_revision, record + ): + break + if self._emit_change: + await self._emit_change( + PrincipalChange( + record.principal_id, + identity.subject, + changed_at or self._now(), + "reporting_subscription_changed", + subscription_id=subscription_id, + ) + ) + return True + + async def reporting_subscriptions( + self, identity: PrincipalIdentity + ) -> list[ReportingSubscriptionState]: + """The caller's stored subscription states, for the reporting runtime.""" + record = await self._store.get(identity.subject) + if record is None or record.configuration is None: + return [] + return [ + item.model_copy(deep=True) + for item in record.configuration.reporting_subscriptions or [] + ] + + async def reporting_subscription_ids(self, identity: PrincipalIdentity) -> set[str]: + """Every subscription_id the caller holds, inactive ones included. + + ``sync_accounts`` must reject a ``delivery_config_id`` equal to one of + them; subscription_id and delivery_config_id share one namespace. + """ + return {item.subscription_id for item in await self.reporting_subscriptions(identity)} + + async def reporting_destinations( + self, identity: PrincipalIdentity + ) -> list[AgentReportingDestinationState]: + """The caller's stored destination states, for the reporting runtime.""" + record = await self._store.get(identity.subject) + if record is None or record.configuration is None: + return [] + return [ + item.model_copy(deep=True) for item in record.configuration.reporting_destinations or [] + ] + def _negotiate_declarations(self, declared: AgentDeclarations) -> PrincipalDeclarationsState: raw = declared.model_dump(mode="json", exclude_none=True) accepted: dict[str, list[str]] = {} @@ -836,17 +1106,15 @@ def _negotiate_declarations(self, declared: AgentDeclarations) -> PrincipalDecla } ) - @staticmethod - def _submitted_sections_are_empty(configuration: PrincipalConfiguration) -> bool: + def _submitted_sections_are_empty(self, configuration: PrincipalConfiguration) -> bool: fields = configuration.model_fields_set if not fields: return False + sections = {"notification_configs", "reporting_destinations"} + if self._subscriptions is not None: + sections.add("reporting_subscriptions") return ( - all( - getattr(configuration, field) == [] - for field in fields - if field in {"notification_configs", "reporting_destinations"} - ) + all(getattr(configuration, field) == [] for field in fields if field in sections) and "declarations" not in fields ) @@ -880,6 +1148,7 @@ def _current_response( __all__ = [ + "DeliveryConfigIdsHook", "DestinationProofHook", "InMemoryPrincipalRecordStore", "NotificationProofHook", @@ -889,4 +1158,6 @@ def _current_response( "PrincipalRecord", "PrincipalRecordStore", "PrincipalService", + "ReportingSubscriptionOptions", + "SubscriptionVersionsHook", ] diff --git a/tests/test_principal_reporting_subscriptions.py b/tests/test_principal_reporting_subscriptions.py new file mode 100644 index 000000000..e8375dfaf --- /dev/null +++ b/tests/test_principal_reporting_subscriptions.py @@ -0,0 +1,721 @@ +"""sync_principal / get_principal ``reporting_subscriptions`` (AdCP 3.3, experimental).""" + +from __future__ import annotations + +from copy import deepcopy +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest + +from adcp.server import ( + PrincipalChange, + PrincipalIdentity, + PrincipalService, + ReportingSubscriptionOptions, +) + +IDENTITY = PrincipalIdentity("https://buyer.example/agent", "buyer_agent") +BETA_2 = SimpleNamespace(resolved_adcp_version="3.3-beta.2") +T0 = datetime(2026, 10, 1, 12, 0, tzinfo=timezone.utc) +DESTINATION = { + "pattern": "file_transfer", + "destination_id": "archive", + "active": True, + "provider": {"domain": "object-store.example"}, + "transport": "s3", + "location": "s3://buyer-reporting/adcp/", + "accepted_formats": ["parquet"], + "accepted_verification_profiles": ["manifest_checksums"], +} +WAREHOUSE = {**DESTINATION, "destination_id": "warehouse", "location": "s3://buyer-wh/adcp/"} + + +def subscription(subscription_id: str = "analytics-daily", **overrides) -> dict: + payload = { + "subscription_id": subscription_id, + "subscription_version": 1, + "active": True, + "scope": {"all_authorized_accounts": True}, + "selector": { + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "dimensions": ["package"], + }, + "reconciliation_mode": "delivery_only", + } + payload.update(overrides) + return payload + + +class Clock: + def __init__(self) -> None: + self.now = T0 + + def __call__(self) -> datetime: + return self.now + + def advance(self, **kwargs) -> None: + self.now += timedelta(**kwargs) + + +def service(*, delivery_config_ids: tuple[str, ...] = (), emitted: list | None = None): + clock = Clock() + + async def ids(_identity: PrincipalIdentity) -> tuple[str, ...]: + return delivery_config_ids + + async def emit(change: PrincipalChange) -> None: + if emitted is not None: + emitted.append(change) + + principals = PrincipalService( + emit_change=emit, + reporting_subscriptions=ReportingSubscriptionOptions(delivery_config_ids=ids, clock=clock), + ) + return principals, clock + + +_keys = iter(range(10_000)) + + +async def sync(principals: PrincipalService, configuration: dict, *, context=None, **extra): + return await principals.sync_principal( + IDENTITY, + { + "idempotency_key": f"subscriptions-key-{next(_keys):06d}", + "configuration": configuration, + **extra, + }, + context=context, + ) + + +def states(response) -> dict: + configuration = response.result.configuration + return { + item.subscription_id: item.model_dump(mode="json", exclude_none=True) + for item in configuration.reporting_subscriptions or [] + } + + +def error_code(response) -> str: + assert response.result.kind == "failed", response + return response.result.errors[0].code + + +# -- option off -------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_without_the_option_the_section_is_rejected() -> None: + # Unsupported sections are rejected, not ignored (sync_principal.mdx): a + # buyer told "applied" would believe it registered and receive nothing. + principals = PrincipalService() + response = await sync(principals, {"reporting_subscriptions": [subscription()]}) + assert error_code(response) == "UNSUPPORTED_FEATURE" + assert (await principals.get_principal(IDENTITY)).result.kind == "unconfigured" + # The other sections behave exactly as before. + other = await sync(principals, {"reporting_destinations": [DESTINATION]}) + assert other.result.kind == "applied" + + +# -- acceptance and readback ----------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_accepts_and_reads_back_state() -> None: + principals, _clock = service() + response = await sync( + principals, + { + "reporting_destinations": [DESTINATION], + "reporting_subscriptions": [ + subscription( + destination={ + "destination_id": "archive", + "path_template": "analytics/{account_id}/{period_start}", + } + ) + ], + }, + ) + assert response.result.kind == "applied" + state = states(response)["analytics-daily"] + destination_ref = response.result.configuration.reporting_destinations[0].destination_ref + assert state == { + "subscription_id": "analytics-daily", + "subscription_version": 1, + "state": "active", + "accepted_at": "2026-10-01T12:00:00Z", + "destination_ref": destination_ref, + "configuration": subscription( + destination={ + "destination_id": "archive", + "path_template": "analytics/{account_id}/{period_start}", + } + ), + } + current = await principals.get_principal(IDENTITY) + assert current.result.configuration.reporting_subscriptions[0].accepted_at == T0 + + +@pytest.mark.asyncio +async def test_identical_resend_is_unchanged_and_keeps_accepted_at() -> None: + principals, clock = service() + first = await sync(principals, {"reporting_subscriptions": [subscription()]}) + clock.advance(hours=3) + again = await sync(principals, {"reporting_subscriptions": [subscription()]}) + assert again.result.action == "unchanged" + assert again.result.configuration_version == first.result.configuration_version + assert states(again)["analytics-daily"]["accepted_at"] == "2026-10-01T12:00:00Z" + + +# -- versioning -------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_new_version_records_prior_versions() -> None: + principals, clock = service() + await sync(principals, {"reporting_subscriptions": [subscription()]}) + clock.advance(days=2) + v2 = subscription(subscription_version=2, exclude_account_ids=["acct-9"]) + response = await sync(principals, {"reporting_subscriptions": [v2]}) + state = states(response)["analytics-daily"] + assert state["subscription_version"] == 2 + assert state["accepted_at"] == "2026-10-03T12:00:00Z" + assert state["prior_versions"] == [ + { + "subscription_version": 1, + "accepted_at": "2026-10-01T12:00:00Z", + "superseded_at": "2026-10-03T12:00:00Z", + } + ] + clock.advance(days=1) + v3 = subscription(subscription_version=3) + state = states(await sync(principals, {"reporting_subscriptions": [v3]}))["analytics-daily"] + assert [item["subscription_version"] for item in state["prior_versions"]] == [2, 1] + + +@pytest.mark.asyncio +async def test_reused_version_with_new_semantics_is_rejected() -> None: + principals, _clock = service() + await sync(principals, {"reporting_subscriptions": [subscription()]}) + changed = subscription(exclude_account_ids=["acct-9"]) + assert error_code(await sync(principals, {"reporting_subscriptions": [changed]})) == ( + "INVALID_REQUEST" + ) + older = subscription(subscription_version=1) + await sync(principals, {"reporting_subscriptions": [subscription(subscription_version=2)]}) + assert error_code(await sync(principals, {"reporting_subscriptions": [older]})) == ( + "INVALID_REQUEST" + ) + + +@pytest.mark.asyncio +async def test_active_changes_in_place() -> None: + principals, clock = service() + await sync(principals, {"reporting_subscriptions": [subscription()]}) + clock.advance(hours=5) + response = await sync(principals, {"reporting_subscriptions": [subscription(active=False)]}) + state = states(response)["analytics-daily"] + assert state["subscription_version"] == 1 + assert state["state"] == "inactive" + assert state["deactivated_at"] == "2026-10-01T17:00:00Z" + assert state["accepted_at"] == "2026-10-01T12:00:00Z" + clock.advance(hours=1) + again = await sync(principals, {"reporting_subscriptions": [subscription(active=False)]}) + assert states(again)["analytics-daily"]["deactivated_at"] == "2026-10-01T17:00:00Z" + + +@pytest.mark.asyncio +async def test_replace_account_configs_is_refused_until_implemented() -> None: + principals, _clock = service() + response = await sync( + principals, {"reporting_subscriptions": [subscription(replace_account_configs=True)]} + ) + assert error_code(response) == "UNSUPPORTED_FEATURE" + allowed = await sync( + principals, {"reporting_subscriptions": [subscription(replace_account_configs=False)]} + ) + assert allowed.result.kind == "applied" + + +@pytest.mark.asyncio +async def test_readback_lists_inactive_intervals() -> None: + principals, clock = service() + await sync(principals, {"reporting_subscriptions": [subscription()]}) + clock.advance(hours=5) + await sync(principals, {"reporting_subscriptions": [subscription(active=False)]}) + clock.advance(hours=24) + reactivated = await sync( + principals, {"reporting_subscriptions": [subscription()]}, context=BETA_2 + ) + state = states(reactivated)["analytics-daily"] + assert state["state"] == "active" and "deactivated_at" not in state + assert state["inactive_intervals"] == [ + {"deactivated_at": "2026-10-01T17:00:00Z", "reactivated_at": "2026-10-02T17:00:00Z"} + ] + clock.advance(hours=1) + omitted = await sync(principals, {"reporting_subscriptions": []}, context=BETA_2) + state = states(omitted)["analytics-daily"] + assert state["inactive_intervals"][-1] == {"deactivated_at": "2026-10-02T18:00:00Z"} + assert state["deactivated_at"] == "2026-10-02T18:00:00Z" + current = await principals.get_principal(IDENTITY, context=BETA_2) + read = current.result.configuration.reporting_subscriptions[0] + assert len(read.inactive_intervals) == 2 + # A caller served 3.3.0-beta.1 (or with no known release) never sees the + # field, so the response validates against the pinned schema. + for context in (None, SimpleNamespace(resolved_adcp_version="3.3-beta.1")): + older = await principals.get_principal(IDENTITY, context=context) + assert older.result.configuration.reporting_subscriptions[0].inactive_intervals is None + + +@pytest.mark.asyncio +async def test_omitted_subscription_is_deactivated_and_retained() -> None: + principals, clock = service() + await sync( + principals, + { + "reporting_subscriptions": [ + subscription(), + subscription( + "pacing", + selector={ + "feed_purpose": "pacing", + "period_duration": "PT1H", + "required_finality": "snapshot", + }, + ), + ] + }, + ) + clock.advance(hours=2) + response = await sync(principals, {"reporting_subscriptions": [subscription()]}) + pacing = states(response)["pacing"] + assert pacing["state"] == "inactive" + assert pacing["deactivated_at"] == "2026-10-01T14:00:00Z" + assert pacing["configuration"]["active"] is False + cleared = await sync(principals, {"reporting_subscriptions": []}) + assert cleared.result.action == "cleared" + assert {item["state"] for item in states(cleared).values()} == {"inactive"} + + +# -- validation ------------------------------------------------------------------------------ + + +INVALID = { + "both scopes": subscription(scope={"all_authorized_accounts": True, "account_ids": ["a"]}), + "empty scope": subscription(scope={}), + "exclude with account_ids": subscription( + scope={"account_ids": ["a"]}, exclude_account_ids=["b"] + ), + "period_timezone with all accounts": subscription( + selector={ + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "period_timezone": "America/New_York", + } + ), + "period_timezone with two accounts": subscription( + scope={"account_ids": ["a", "b"]}, + selector={ + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "period_timezone": "America/New_York", + }, + ), + "unknown period_timezone": subscription( + scope={"account_ids": ["a"]}, + selector={ + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "period_timezone": "Mars/Olympus_Mons", + }, + ), + "billing with snapshot": subscription( + selector={ + "feed_purpose": "billing", + "period_duration": "P1D", + "required_finality": "snapshot", + }, + reconciliation_mode="consumer_receipt", + destination={"destination_id": "archive", "path_template": "b/{account_id}"}, + ), + "billing without receipts": subscription( + selector={ + "feed_purpose": "billing", + "period_duration": "P1D", + "required_finality": "official", + }, + destination={"destination_id": "archive", "path_template": "b/{account_id}"}, + ), + "receipts without destination": subscription(reconciliation_mode="consumer_receipt"), + "missing destination": subscription( + destination={"destination_id": "nowhere", "path_template": "x/{account_id}"} + ), + "multi-account file_transfer without template": subscription( + destination={"destination_id": "archive"} + ), + "multi-account template without account_id": subscription( + destination={"destination_id": "archive", "path_template": "analytics/{period_start}"} + ), + "unknown placeholder": subscription( + destination={"destination_id": "archive", "path_template": "a/{account_id}/{brand}"} + ), + "subscription_id equals a delivery_config_id": subscription("acme-pacing"), +} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("case", sorted(INVALID)) +async def test_invalid_subscriptions_fail_the_whole_request(case: str) -> None: + principals, _clock = service(delivery_config_ids=("acme-pacing",)) + response = await sync( + principals, + {"reporting_destinations": [DESTINATION], "reporting_subscriptions": [INVALID[case]]}, + ) + assert error_code(response) == "INVALID_REQUEST" + assert (await principals.get_principal(IDENTITY)).result.kind == "unconfigured" + + +@pytest.mark.asyncio +async def test_single_account_subscription_needs_no_account_placeholder() -> None: + principals, _clock = service() + single = subscription( + scope={"account_ids": ["acct-1"]}, + destination={"destination_id": "archive", "path_template": "acme/{period_start}"}, + selector={ + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "period_timezone": "America/Chicago", + }, + ) + response = await sync( + principals, {"reporting_destinations": [DESTINATION], "reporting_subscriptions": [single]} + ) + assert response.result.kind == "applied" + + +@pytest.mark.asyncio +async def test_set_level_rules() -> None: + principals, _clock = service() + duplicate = await sync( + principals, {"reporting_subscriptions": [subscription(), subscription()]} + ) + assert error_code(duplicate) == "INVALID_REQUEST" + two_all = await sync( + principals, + {"reporting_subscriptions": [subscription("a"), subscription("b")]}, + ) + assert error_code(two_all) == "INVALID_REQUEST" + disjoint_operators = await sync( + principals, + { + "reporting_subscriptions": [ + subscription("a", operator_domains=["one.example"]), + subscription("b", operator_domains=["two.example"]), + ] + }, + ) + assert disjoint_operators.result.kind == "applied" + shared_accounts = await sync( + principals, + { + "reporting_subscriptions": [ + subscription("a", scope={"account_ids": ["x", "y"]}), + subscription("b", scope={"account_ids": ["y"]}), + ] + }, + ) + assert error_code(shared_accounts) == "INVALID_REQUEST" + # Different levels are precedence, not a conflict. + levels = await sync( + principals, + { + "reporting_subscriptions": [ + subscription("all"), + subscription("named", scope={"account_ids": ["y"]}), + ] + }, + ) + assert levels.result.kind == "applied" + + +@pytest.mark.asyncio +async def test_shared_destination_templates_must_differ() -> None: + principals, _clock = service() + first = subscription( + "a", destination={"destination_id": "archive", "path_template": "x/{account_id}"} + ) + second = subscription( + "b", + selector={ + "feed_purpose": "pacing", + "period_duration": "PT1H", + "required_finality": "snapshot", + }, + destination={"destination_id": "archive", "path_template": "x/{account_id}"}, + ) + response = await sync( + principals, + {"reporting_destinations": [DESTINATION], "reporting_subscriptions": [first, second]}, + ) + assert error_code(response) == "INVALID_REQUEST" + second["destination"]["path_template"] = "y/{account_id}" + response = await sync( + principals, + {"reporting_destinations": [DESTINATION], "reporting_subscriptions": [first, second]}, + ) + assert response.result.kind == "applied" + + +# -- interaction with the other sections --------------------------------------------------- + + +@pytest.mark.asyncio +async def test_destination_removed_by_a_later_request_fails_it() -> None: + principals, _clock = service() + named = subscription( + destination={"destination_id": "warehouse", "path_template": "a/{account_id}"} + ) + await sync( + principals, + {"reporting_destinations": [DESTINATION, WAREHOUSE], "reporting_subscriptions": [named]}, + ) + response = await sync(principals, {"reporting_destinations": [DESTINATION]}) + assert error_code(response) == "INVALID_REQUEST" + # Removing both in one request is fine: the subscription is deactivated. + both = await sync( + principals, {"reporting_destinations": [DESTINATION], "reporting_subscriptions": []} + ) + assert both.result.kind == "applied" + + +@pytest.mark.asyncio +async def test_dry_run_and_stale_configuration_version() -> None: + principals, _clock = service() + first = await sync(principals, {"reporting_subscriptions": [subscription()]}) + dry = await sync( + principals, + {"reporting_subscriptions": [subscription(subscription_version=2)]}, + dry_run=True, + ) + assert dry.result.kind == "validated" + assert dry.result.action == "would_update" + current = await principals.get_principal(IDENTITY) + assert current.result.configuration.reporting_subscriptions[0].subscription_version == 1 + invalid_dry = await sync( + principals, {"reporting_subscriptions": [subscription(scope={})]}, dry_run=True + ) + assert error_code(invalid_dry) == "INVALID_REQUEST" + stale = await sync( + principals, + {"reporting_subscriptions": [subscription(subscription_version=2)]}, + expected_configuration_version="cfg:stale", + ) + assert error_code(stale) == "CONFLICT" + fresh = await sync( + principals, + {"reporting_subscriptions": [subscription(subscription_version=2)]}, + expected_configuration_version=first.result.configuration_version, + ) + assert fresh.result.kind == "applied" + + +@pytest.mark.asyncio +async def test_destination_needing_action_moves_the_subscription_to_action_required() -> None: + emitted: list[PrincipalChange] = [] + principals, _clock = service(emitted=emitted) + named = subscription( + destination={"destination_id": "archive", "path_template": "a/{account_id}"} + ) + await sync( + principals, {"reporting_destinations": [DESTINATION], "reporting_subscriptions": [named]} + ) + await principals.transition_destination( + IDENTITY, + "archive", + "action_required", + issues=[{"code": "ACCESS_DENIED", "message": "grant write access"}], + ) + state = (await principals.get_principal(IDENTITY)).result.configuration + subscription_state = state.reporting_subscriptions[0] + assert subscription_state.state == "action_required" + assert subscription_state.issues[0].code == "DESTINATION_NOT_READY" + reasons = [(item.reason, item.subscription_id) for item in emitted] + assert ("reporting_subscription_changed", "analytics-daily") in reasons + emitted.clear() + await principals.transition_destination(IDENTITY, "archive", "ready") + state = (await principals.get_principal(IDENTITY)).result.configuration + assert state.reporting_subscriptions[0].state == "active" + assert state.reporting_subscriptions[0].issues is None + # The recovery is announced too. + assert ("reporting_subscription_changed", "analytics-daily") in [ + (item.reason, item.subscription_id) for item in emitted + ] + + +@pytest.mark.asyncio +async def test_destination_transitions_work_without_a_change_emitter() -> None: + principals = PrincipalService(reporting_subscriptions=ReportingSubscriptionOptions()) + named = subscription( + destination={"destination_id": "archive", "path_template": "a/{account_id}"} + ) + await sync( + principals, {"reporting_destinations": [DESTINATION], "reporting_subscriptions": [named]} + ) + await principals.transition_destination(IDENTITY, "archive", "action_required") + await principals.transition_destination(IDENTITY, "archive", "ready") + state = (await principals.get_principal(IDENTITY)).result.configuration + assert state.reporting_subscriptions[0].state == "active" + + +@pytest.mark.asyncio +async def test_seller_status_updates_do_not_bump_the_configuration_version() -> None: + emitted: list[PrincipalChange] = [] + principals, _clock = service(emitted=emitted) + first = await sync(principals, {"reporting_subscriptions": [subscription()]}) + changed = await principals.update_subscription_status( + IDENTITY, + "analytics-daily", + accounts_with_exceptions=[{"account_id": "acct-1", "reasons": ["capacity"]}], + accounts_unauthorized=[], + ) + assert changed is True + unchanged = await principals.update_subscription_status( + IDENTITY, + "analytics-daily", + accounts_with_exceptions=[{"account_id": "acct-1", "reasons": ["capacity"]}], + ) + assert unchanged is False + current = await principals.get_principal(IDENTITY) + assert current.result.configuration_version == first.result.configuration_version + state = current.result.configuration.reporting_subscriptions[0] + assert state.accounts_with_exceptions[0].reasons[0] == "capacity" + assert [item.reason for item in emitted] == ["reporting_subscription_changed"] + # A resend keeps seller-derived status. + resend = await sync(principals, {"reporting_subscriptions": [deepcopy(subscription())]}) + assert states(resend)["analytics-daily"]["accounts_with_exceptions"] == [ + {"account_id": "acct-1", "reasons": ["capacity"]} + ] + + +@pytest.mark.asyncio +async def test_resends_that_reorder_or_spell_out_defaults_are_the_same_version() -> None: + principals, _clock = service() + first = subscription(scope={"account_ids": ["b", "a"]}) + await sync(principals, {"reporting_subscriptions": [first]}) + reordered = subscription( + scope={"account_ids": ["a", "b"]}, + selector={**first["selector"], "match": "exact"}, + ) + response = await sync(principals, {"reporting_subscriptions": [reordered]}) + assert response.result.kind == "applied" + + +@pytest.mark.asyncio +async def test_braces_outside_placeholders_are_refused() -> None: + principals, _clock = service() + for template in ("a/{{account_id}}", "a/}{account_id}{", "a/{account_id}/x}"): + response = await sync( + principals, + { + "reporting_destinations": [DESTINATION], + "reporting_subscriptions": [ + subscription( + destination={"destination_id": "archive", "path_template": template} + ) + ], + }, + ) + assert error_code(response) == "INVALID_REQUEST", template + + +def test_render_path_template() -> None: + from adcp.reporting.subscriptions import render_path_template + + rendered = render_path_template( + "analytics/{account_id}/{offering_id}/{period_start}", + account_id="acme/eu", + subscription_id="analytics-daily", + offering_id="..", + period_start=datetime(2026, 10, 1, 4, tzinfo=timezone.utc), + ) + # A value with "/" or equal to "." / ".." fills exactly one segment. + assert rendered == "analytics/acme%2Feu/%2E%2E/2026-10-01T04:00:00Z" + backslash = render_path_template( + "a/{account_id}/{subscription_id}", + account_id="..\\..\\secret", + subscription_id="analytics-daily", + offering_id="o", + period_start=datetime(2026, 10, 1, tzinfo=timezone.utc), + ) + assert backslash == "a/..%5C..%5Csecret/analytics-daily" + + +@pytest.mark.asyncio +async def test_inactive_intervals_are_never_dropped() -> None: + principals, clock = service() + await sync(principals, {"reporting_subscriptions": [subscription()]}) + for _ in range(64): + clock.advance(hours=1) + await sync(principals, {"reporting_subscriptions": [subscription(active=False)]}) + clock.advance(hours=1) + await sync(principals, {"reporting_subscriptions": [subscription()]}) + clock.advance(hours=1) + response = await sync(principals, {"reporting_subscriptions": [subscription(active=False)]}) + assert error_code(response) == "INVALID_REQUEST" + current = await principals.get_principal(IDENTITY, context=BETA_2) + assert len(current.result.configuration.reporting_subscriptions[0].inactive_intervals) == 64 + + +@pytest.mark.asyncio +async def test_prior_versions_carry_their_configuration_for_beta_2() -> None: + principals, clock = service() + await sync(principals, {"reporting_subscriptions": [subscription()]}) + clock.advance(days=1) + await sync( + principals, + { + "reporting_subscriptions": [ + subscription(subscription_version=2, exclude_account_ids=["x"]) + ] + }, + ) + served = await principals.get_principal(IDENTITY, context=BETA_2) + (prior,) = served.result.configuration.reporting_subscriptions[0].prior_versions + assert prior.configuration.subscription_version == 1 + older = await principals.get_principal(IDENTITY) + assert ( + older.result.configuration.reporting_subscriptions[0].prior_versions[0].configuration + is None + ) + + +@pytest.mark.asyncio +async def test_subscription_ids_for_the_reverse_namespace_check() -> None: + principals, _clock = service() + await sync( + principals, + { + "reporting_subscriptions": [ + subscription(), + subscription( + "pacing", + selector={ + "feed_purpose": "pacing", + "period_duration": "PT1H", + "required_finality": "snapshot", + }, + ), + ] + }, + ) + await sync(principals, {"reporting_subscriptions": [subscription()]}) + assert await principals.reporting_subscription_ids(IDENTITY) == {"analytics-daily", "pacing"} diff --git a/tests/test_reporting_subscription_engine.py b/tests/test_reporting_subscription_engine.py new file mode 100644 index 000000000..7ad8aae4f --- /dev/null +++ b/tests/test_reporting_subscription_engine.py @@ -0,0 +1,1121 @@ +"""Seller runtime for reporting subscriptions, and buyer/seller conformance. + +The conformance cases follow adcontextprotocol/adcp-client#3504: the buyer +computes its expected periods with ``expected_subscription_periods`` from what +it can see (the ``get_principal`` readback, ``subscription_coverage[]``, its own +buys and grant history), and the seller's owed periods must equal them period +by period. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from types import SimpleNamespace + +import pytest + +from adcp.reporting.subscriptions import ( + AccountDeliveryConfig, + AuthorizedAccount, + AuthorizedAccountPage, + ReportingSubscriptionEngine, + ReportingSubscriptionSupport, + SubscriptionAuthorization, + SubscriptionBuy, + SubscriptionBuyFlight, + SubscriptionBuyPackage, + SubscriptionOffering, + expected_subscription_periods, +) +from adcp.server import PrincipalChange, PrincipalIdentity, PrincipalService + +IDENTITY = PrincipalIdentity("https://agency.example/agent", "buyer_agent") +#: A handler context served 3.3.0-beta.2, which carries the adcp#8585 fields. +BETA_2 = SimpleNamespace(resolved_adcp_version="3.3-beta.2") +NY = "America/New_York" + + +def at(text: str) -> datetime: + return datetime.fromisoformat(text.replace("Z", "+00:00")) + + +def offering(offering_id: str, dimensions: tuple[str, ...], **kwargs) -> SubscriptionOffering: + defaults = dict( + report_definition_id=f"def-{offering_id}", + reporting_profile="delivery", + feed_purpose="analytics", + period_duration="P1D", + alignment="account_timezone", + delivery_sla="PT6H", + supported_finality=frozenset({"official"}), + reconciliation_mode="delivery_only", + ) + defaults.update(kwargs) + return SubscriptionOffering( + offering_id=offering_id, dimensions=frozenset(dimensions), **defaults + ) + + +PACKAGE = offering("pkg", ("package",)) +PACKAGE_CREATIVE = offering("pkg-creative", ("package", "creative")) +PACKAGE_COUNTRY = offering("pkg-country", ("package", "country")) +HOURLY = offering( + "pkg-hourly", ("package",), period_duration="PT1H", alignment="utc", delivery_sla="PT2H" +) +REGION_UTC = offering("pkg-region-utc", ("package", "region"), alignment="utc") +OFFERINGS = [PACKAGE, PACKAGE_CREATIVE, PACKAGE_COUNTRY, HOURLY, REGION_UTC] +ALL_IDS = tuple(item.offering_id for item in OFFERINGS) + + +def subscription(subscription_id: str = "analytics", **overrides) -> dict: + payload = { + "subscription_id": subscription_id, + "subscription_version": 1, + "active": True, + "scope": {"all_authorized_accounts": True}, + "selector": { + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "dimensions": ["package"], + }, + "reconciliation_mode": "delivery_only", + } + payload.update(overrides) + return payload + + +def buy( + media_buy_id: str, + start: str, + end: str, + *, + confirmed: str | None = None, + offering_ids: tuple[str, ...] = ALL_IDS, + **kwargs, +) -> SubscriptionBuy: + return SubscriptionBuy( + media_buy_id=media_buy_id, + confirmed_at=at(confirmed or start), + start_time=at(start), + end_time=at(end), + reporting_delivery_offering_ids=offering_ids, + **kwargs, + ) + + +@dataclass +class Host: + accounts: dict[str, AuthorizedAccount] = field(default_factory=dict) + buys: dict[str, list[SubscriptionBuy]] = field(default_factory=dict) + configs: dict[str, list[AccountDeliveryConfig]] = field(default_factory=dict) + page_size: int = 2 + + async def list_authorized_accounts(self, principal, *, cursor): + ids = sorted(self.accounts) + start = int(cursor or 0) + page = ids[start : start + self.page_size] + following = start + self.page_size + return AuthorizedAccountPage( + accounts=tuple(self.accounts[item] for item in page), + next_cursor=str(following) if following < len(ids) else None, + ) + + async def account_buys(self, principal, account_id): + return list(self.buys.get(account_id, ())) + + async def account_delivery_configs(self, principal, account_id): + return list(self.configs.get(account_id, ())) + + +class Clock: + def __init__(self, start: str) -> None: + self.now = at(start) + + def __call__(self) -> datetime: + return self.now + + def set(self, text: str) -> None: + self.now = at(text) + + +class World: + """A seller (principal service + engine) and the buyer's own records.""" + + def __init__( + self, + *, + start: str = "2026-10-01T00:00:00Z", + support: ReportingSubscriptionSupport | None = None, + retention_days: int = 400, + ) -> None: + self.clock = Clock(start) + self.host = Host() + self.changes: list[PrincipalChange] = [] + + async def emit(change: PrincipalChange) -> None: + self.changes.append(change) + + self.principals = PrincipalService(emit_change=emit) + self.engine = ReportingSubscriptionEngine( + principals=self.principals, + host=self.host, + offerings=OFFERINGS, + support=support, + status_retention_days=retention_days, + clock=self.clock, + ) + self.principals.set_reporting_subscriptions(self.engine.principal_options()) + self.retention_days = retention_days + self.sent: dict[str, dict[int, dict]] = {} + self.grants: dict[str, list[SubscriptionAuthorization]] = {} + self._key = 0 + + async def sync(self, *subscriptions: dict) -> None: + self._key += 1 + response = await self.principals.sync_principal( + IDENTITY, + { + "idempotency_key": f"engine-test-key-{self._key:06d}", + "configuration": {"reporting_subscriptions": list(subscriptions)}, + }, + ) + assert response.result.kind == "applied", response + for item in subscriptions: + self.sent.setdefault(item["subscription_id"], {})[item["subscription_version"]] = item + + def grant(self, account_id: str, when: str, **kwargs) -> None: + self.host.accounts[account_id] = AuthorizedAccount( + account_id, timezone=kwargs.pop("timezone", NY), authorized_at=at(when), **kwargs + ) + history = self.grants.setdefault(account_id, []) + history.append(SubscriptionAuthorization(at(when))) + + def revoke(self, account_id: str, when: str) -> None: + del self.host.accounts[account_id] + history = self.grants[account_id] + history[-1] = SubscriptionAuthorization(history[-1].authorized_at, at(when)) + + async def reconcile(self, when: str) -> None: + self.clock.set(when) + await self.engine.reconcile(IDENTITY) + + async def state(self, subscription_id: str): + current = await self.principals.get_principal(IDENTITY, context=BETA_2) + return next( + item + for item in current.result.configuration.reporting_subscriptions + if item.subscription_id == subscription_id + ) + + async def coverage(self, account_id: str, subscription_id: str): + entries = await self.engine.subscription_coverage( + IDENTITY, account_id, adcp_version="3.3-beta.2" + ) + return next((item for item in entries if item.subscription_id == subscription_id), None) + + async def seller_owes(self, subscription_id: str, account_id: str, now: str): + return await self.engine.owed_periods(IDENTITY, subscription_id, account_id, now=at(now)) + + async def buyer_expects(self, subscription_id: str, account_id: str, now: str, coverage=None): + """What the buyer computes from what it can see, never from engine internals.""" + state = await self.state(subscription_id) + excluded = coverage.excluded if coverage is not None else () + gaps = (coverage.gaps or ()) if coverage is not None else () + return expected_subscription_periods( + subscription=state, + version_configurations=self.sent[subscription_id], + account_id=account_id, + buys=self.host.buys.get(account_id, ()), + offerings=OFFERINGS, + account_timezone=NY, + now=at(now), + authorizations=self.grants.get(account_id, ()), + exceptions=excluded, + gaps=gaps, + backfill_on_regrant=self.engine.support.backfill_on_regrant, + status_retention_days=self.retention_days, + ) + + async def assert_conformant( + self, subscription_id: str, account_id: str, now: str, coverage=None + ) -> list: + seller = await self.seller_owes(subscription_id, account_id, now) + buyer = await self.buyer_expects(subscription_id, account_id, now, coverage) + assert buyer == seller + return seller + + +def table(periods) -> list[tuple]: + return [ + (p.period_start, p.offering_id, p.delivery_config_version, p.media_buy_ids) for p in periods + ] + + +# -- conformance cases ------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_mid_period_confirmation_cancellation_and_completion() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [ + buy( + "mb-confirmed-late", + "2026-10-02T04:00:00Z", + "2026-10-08T04:00:00Z", + confirmed="2026-10-03T15:00:00Z", + ), + buy( + "mb-canceled", + "2026-10-02T04:00:00Z", + "2026-10-08T04:00:00Z", + canceled_at=at("2026-10-04T16:00:00Z"), + ), + buy( + "mb-completed", + "2026-10-02T04:00:00Z", + "2026-10-08T04:00:00Z", + completed_at=at("2026-10-03T01:00:00Z"), + ), + ] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-09T00:00:00Z") + assert table(owed) == [ + ("2026-10-02T04:00:00Z", "pkg", 1, ("mb-canceled", "mb-completed")), + # adcp#8593: the day containing confirmed_at (10-03T15Z) is owed in full. + ("2026-10-03T04:00:00Z", "pkg", 1, ("mb-canceled", "mb-confirmed-late")), + ("2026-10-04T04:00:00Z", "pkg", 1, ("mb-canceled", "mb-confirmed-late")), + ("2026-10-05T04:00:00Z", "pkg", 1, ("mb-confirmed-late",)), + ("2026-10-06T04:00:00Z", "pkg", 1, ("mb-confirmed-late",)), + ("2026-10-07T04:00:00Z", "pkg", 1, ("mb-confirmed-late",)), + ] + + +@pytest.mark.asyncio +async def test_flight_extension() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [ + buy( + "mb-1", + "2026-10-02T04:00:00Z", + "2026-10-09T04:00:00Z", + superseded_flights=( + SubscriptionBuyFlight( + at("2026-10-06T12:00:00Z"), + at("2026-10-02T04:00:00Z"), + at("2026-10-04T04:00:00Z"), + ), + ), + ), + ] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-10T00:00:00Z") + assert [p.period_start for p in owed] == [ + "2026-10-02T04:00:00Z", + "2026-10-03T04:00:00Z", + "2026-10-06T04:00:00Z", + "2026-10-07T04:00:00Z", + "2026-10-08T04:00:00Z", + ] + + +@pytest.mark.asyncio +async def test_new_subscription_version() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-06T04:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.clock.set("2026-10-03T09:30:00Z") + v2 = subscription( + subscription_version=2, + selector={ + "feed_purpose": "analytics", + "period_duration": "PT1H", + "required_finality": "official", + "dimensions": ["package"], + }, + ) + await world.sync(v2) + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-04T07:00:00Z") + daily = [(p.period_start, p.delivery_config_version) for p in owed if p.offering_id == "pkg"] + hourly = [p.period_start for p in owed if p.offering_id == "pkg-hourly"] + assert daily == [ + ("2026-10-01T04:00:00Z", 1), + ("2026-10-02T04:00:00Z", 1), + ("2026-10-03T04:00:00Z", 1), + ] + # v2 starts at E, the end of the Oct 3 local day. + assert hourly[0] == "2026-10-04T04:00:00Z" + assert len(hourly) == 3 + + +@pytest.mark.asyncio +async def test_revocation_and_regrant_without_backfill() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-12T04:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + + world.revoke("acct-1", "2026-10-04T08:00:00Z") + await world.reconcile("2026-10-04T08:00:00Z") + state = await world.state("analytics") + entry = state.accounts_unauthorized[0] + assert entry.account_id == "acct-1" + assert entry.unauthorized_at == at("2026-10-04T08:00:00Z") + assert entry.last_owed_period_end == at("2026-10-04T04:00:00Z") + assert entry.withheld_period_count == 1 # the Oct 3 day was not yet due + assert await world.engine.subscription_coverage(IDENTITY, "acct-1") == [] + assert any(c.reason == "reporting_subscription_changed" for c in world.changes) + await world.assert_conformant("analytics", "acct-1", "2026-10-05T00:00:00Z") + + world.grant("acct-1", "2026-10-06T15:00:00Z") + await world.reconcile("2026-10-06T15:00:00Z") + assert (await world.state("analytics")).accounts_unauthorized is None + coverage = await world.coverage("acct-1", "analytics") + assert [(g.from_, g.until) for g in coverage.gaps] == [ + (at("2026-10-04T04:00:00Z"), at("2026-10-07T04:00:00Z")) + ] + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-10T00:00:00Z", coverage) + assert [(p.period_start, p.expected_at) for p in owed] == [ + ("2026-10-01T04:00:00Z", "2026-10-02T10:00:00Z"), + ("2026-10-02T04:00:00Z", "2026-10-03T10:00:00Z"), + ("2026-10-03T04:00:00Z", "2026-10-06T21:00:00Z"), + ("2026-10-07T04:00:00Z", "2026-10-08T10:00:00Z"), + ("2026-10-08T04:00:00Z", "2026-10-09T10:00:00Z"), + ] + + +@pytest.mark.asyncio +async def test_withheld_count_uses_the_delivery_hook_when_present() -> None: + world = World() + delivered: set[str] = {"2026-10-01T04:00:00Z"} + + async def is_period_delivered(principal, account_id, period, *, by): + assert by == at("2026-10-04T08:00:00Z") + return period.period_start in delivered + + world.host.is_period_delivered = is_period_delivered # type: ignore[attr-defined] + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-12T04:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.revoke("acct-1", "2026-10-04T08:00:00Z") + await world.reconcile("2026-10-04T08:00:00Z") + # Oct 2 and Oct 3 ended before the loss and were not delivered. + entry = (await world.state("analytics")).accounts_unauthorized[0] + assert entry.withheld_period_count == 2 + + +@pytest.mark.asyncio +async def test_revocation_and_regrant_with_backfill() -> None: + world = World(support=ReportingSubscriptionSupport(backfill_on_regrant=True)) + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-12T04:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.revoke("acct-1", "2026-10-04T08:00:00Z") + await world.reconcile("2026-10-04T08:00:00Z") + world.grant("acct-1", "2026-10-06T15:00:00Z") + await world.reconcile("2026-10-06T15:00:00Z") + coverage = await world.coverage("acct-1", "analytics") + assert coverage.gaps is None + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-09T00:00:00Z", coverage) + assert [p.period_start for p in owed] == [f"2026-10-{day:02d}T04:00:00Z" for day in range(1, 8)] + + +@pytest.mark.asyncio +async def test_max_accounts_is_deterministic_and_reported() -> None: + world = World(support=ReportingSubscriptionSupport(max_accounts=2)) + world.grant("acct-b", "2026-09-01T00:00:00Z") + world.grant("acct-a", "2026-09-01T00:00:00Z") + world.grant("acct-c", "2026-09-02T00:00:00Z") + for account in ("acct-a", "acct-b", "acct-c"): + world.host.buys[account] = [ + buy(f"mb-{account}", "2026-10-01T04:00:00Z", "2026-10-05T04:00:00Z") + ] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + state = await world.state("analytics") + assert [(e.account_id, e.reasons) for e in state.accounts_with_exceptions] == [ + ("acct-c", ["capacity"]) + ] + coverage = await world.coverage("acct-c", "analytics") + assert coverage.status == "excluded" + assert coverage.excluded[0].reason == "capacity" + assert ( + await world.assert_conformant("analytics", "acct-c", "2026-10-06T00:00:00Z", coverage) == [] + ) + # acct-a leaves: acct-c is served from the next boundary. + world.revoke("acct-a", "2026-10-02T13:00:00Z") + await world.reconcile("2026-10-02T13:00:00Z") + coverage = await world.coverage("acct-c", "analytics") + assert coverage.status == "active" + assert coverage.excluded[0].until == at("2026-10-03T04:00:00Z") + owed = await world.assert_conformant("analytics", "acct-c", "2026-10-06T00:00:00Z", coverage) + assert [p.period_start for p in owed] == ["2026-10-03T04:00:00Z", "2026-10-04T04:00:00Z"] + + +@pytest.mark.asyncio +async def test_max_accounts_slots_are_kept_by_served_accounts() -> None: + # acct-a lost its slot when it lost authority; acct-b, served since, + # keeps it when acct-a comes back. + world = World(support=ReportingSubscriptionSupport(max_accounts=1), retention_days=2) + world.grant("acct-a", "2026-09-01T00:00:00Z") + await world.sync(subscription()) + await world.reconcile("2026-09-01T00:00:00Z") + world.revoke("acct-a", "2026-09-02T00:00:00Z") + await world.reconcile("2026-09-02T00:00:00Z") + world.grant("acct-b", "2026-09-05T00:00:00Z") + await world.reconcile("2026-09-05T00:00:00Z") + world.grant("acct-a", "2026-09-10T00:00:00Z") + await world.reconcile("2026-09-10T00:00:00Z") + state = await world.state("analytics") + assert [(e.account_id, e.reasons) for e in state.accounts_with_exceptions] == [ + ("acct-a", ["capacity"]) + ] + # A byte-lower account granted later does not displace the served one. + world.grant("acct-0", "2026-09-11T00:00:00Z") + await world.reconcile("2026-09-11T00:00:00Z") + state = await world.state("analytics") + assert [(e.account_id, e.reasons) for e in state.accounts_with_exceptions] == [ + ("acct-0", ["capacity"]), + ("acct-a", ["capacity"]), + ] + + +@pytest.mark.asyncio +async def test_precedence_over_an_account_level_configuration() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-08T04:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.host.configs["acct-1"] = [AccountDeliveryConfig("acme-analytics", "analytics")] + await world.reconcile("2026-10-02T10:00:00Z") + coverage = await world.coverage("acct-1", "analytics") + assert coverage.status == "excluded" + entry = coverage.excluded[0] + assert entry.reason == "overridden" + assert entry.overridden_by.delivery_config_id == "acme-analytics" + assert entry.from_ == at("2026-10-03T04:00:00Z") + world.host.configs["acct-1"] = [] + await world.reconcile("2026-10-04T20:00:00Z") + coverage = await world.coverage("acct-1", "analytics") + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-09T00:00:00Z", coverage) + assert [p.period_start for p in owed] == [ + "2026-10-01T04:00:00Z", + "2026-10-02T04:00:00Z", + "2026-10-05T04:00:00Z", + "2026-10-06T04:00:00Z", + "2026-10-07T04:00:00Z", + ] + + +@pytest.mark.asyncio +async def test_account_ids_subscription_beats_all_accounts() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.grant("acct-2", "2026-09-01T00:00:00Z") + for account in ("acct-1", "acct-2"): + world.host.buys[account] = [ + buy(f"mb-{account}", "2026-10-01T04:00:00Z", "2026-10-03T04:00:00Z") + ] + await world.sync(subscription("all"), subscription("named", scope={"account_ids": ["acct-2"]})) + await world.reconcile("2026-10-01T00:00:00Z") + coverage = await world.coverage("acct-2", "all") + assert coverage.excluded[0].reason == "overridden" + assert coverage.excluded[0].overridden_by.subscription_id == "named" + assert await world.assert_conformant("all", "acct-2", "2026-10-04T00:00:00Z", coverage) == [] + named = await world.assert_conformant( + "named", "acct-2", "2026-10-04T00:00:00Z", await world.coverage("acct-2", "named") + ) + assert len(named) == 2 + assert len(await world.assert_conformant("all", "acct-1", "2026-10-04T00:00:00Z")) == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "match, expected", + [ + ("exact", ["pkg"]), + ("finest", ["pkg-country"]), + ("all", ["pkg", "pkg-country", "pkg-creative", "pkg-region-utc"]), + ], +) +async def test_exact_finest_and_all(match: str, expected: list[str]) -> None: + # pkg-country sorts before pkg-creative in byte order, so finest picks it. + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-02T04:00:00Z")] + await world.sync( + subscription( + selector={ + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "dimensions": ["package"], + "match": match, + } + ) + ) + await world.reconcile("2026-10-01T00:00:00Z") + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-03T00:00:00Z") + assert sorted({p.offering_id for p in owed}) == expected + + +@pytest.mark.asyncio +async def test_unsupported_buys_and_packages_and_buyer_exclusions() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.grant("acct-x", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [ + buy( + "mb-none", + "2026-10-01T04:00:00Z", + "2026-10-03T04:00:00Z", + offering_ids=("pkg-creative",), + ), + SubscriptionBuy( + media_buy_id="mb-some", + confirmed_at=at("2026-10-01T00:00:00Z"), + start_time=at("2026-10-01T04:00:00Z"), + end_time=at("2026-10-03T04:00:00Z"), + packages=( + SubscriptionBuyPackage("p1", ("pkg",)), + SubscriptionBuyPackage("p2", ("pkg-creative",)), + ), + ), + ] + await world.sync(subscription(exclude_account_ids=["acct-x"])) + await world.reconcile("2026-10-01T00:00:00Z") + coverage = await world.coverage("acct-1", "analytics") + assert coverage.status == "active" + assert [(e.media_buy_id, e.package_ids) for e in coverage.excluded] == [ + ("mb-none", None), + ("mb-some", ["p2"]), + ] + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-04T00:00:00Z", coverage) + assert {p.media_buy_ids for p in owed} == {("mb-some",)} + excluded = await world.coverage("acct-x", "analytics") + assert excluded.status == "excluded" + assert excluded.excluded[0].reason == "excluded_by_buyer" + state = await world.state("analytics") + assert [(e.account_id, e.reasons) for e in state.accounts_with_exceptions] == [ + ("acct-1", ["offering_unsupported"]), + ("acct-x", ["excluded_by_buyer"]), + ] + + +@pytest.mark.asyncio +async def test_expired_unauthorized_entry_drops_and_regrant_is_a_fresh_start() -> None: + world = World(retention_days=2) + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-12T04:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.revoke("acct-1", "2026-10-03T08:00:00Z") + await world.reconcile("2026-10-03T08:00:00Z") + assert (await world.state("analytics")).accounts_unauthorized + await world.reconcile("2026-10-06T00:00:00Z") + assert (await world.state("analytics")).accounts_unauthorized is None + world.grant("acct-1", "2026-10-07T12:00:00Z") + await world.reconcile("2026-10-07T12:00:00Z") + coverage = await world.coverage("acct-1", "analytics") + assert coverage.gaps is None + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-10T00:00:00Z", coverage) + assert [(p.period_start, p.withheld) for p in owed] == [ + ("2026-10-01T04:00:00Z", False), + ("2026-10-02T04:00:00Z", True), + ("2026-10-08T04:00:00Z", False), + ] + + +@pytest.mark.asyncio +async def test_reactivation_in_place_is_a_fresh_start() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-09T04:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.clock.set("2026-10-03T12:00:00Z") + await world.sync(subscription(active=False)) + world.clock.set("2026-10-05T12:00:00Z") + await world.sync(subscription()) + state = await world.state("analytics") + assert state.state == "active" and state.deactivated_at is None + assert len(state.inactive_intervals) == 1 + # The buyer recomputes the same periods from the readback alone. + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-09T05:00:00Z") + # Oct 3 started before deactivation; the next owed day starts at the + # first boundary after reactivation. + assert [p.period_start for p in owed] == [ + "2026-10-01T04:00:00Z", + "2026-10-02T04:00:00Z", + "2026-10-03T04:00:00Z", + "2026-10-06T04:00:00Z", + "2026-10-07T04:00:00Z", + "2026-10-08T04:00:00Z", + ] + + +@pytest.mark.asyncio +async def test_timezone_cutovers_carry_previous_timezone() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-07T07:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.host.accounts["acct-1"] = AuthorizedAccount( + "acct-1", timezone="America/Los_Angeles", authorized_at=at("2026-09-01T00:00:00Z") + ) + await world.reconcile("2026-10-03T12:00:00Z") + coverage = await world.coverage("acct-1", "analytics") + (cutover,) = coverage.timezone_cutovers + assert cutover.previous_timezone == NY + assert cutover.timezone == "America/Los_Angeles" + assert cutover.effective_at == at("2026-10-04T04:00:00Z") + assert cutover.offering_id is None + # The buyer needs no timezone history of its own: list_accounts shows the + # current zone, and previous_timezone gives the zone before the cutover. + seller = await world.seller_owes("analytics", "acct-1", "2026-10-08T00:00:00Z") + buyer = expected_subscription_periods( + subscription=await world.state("analytics"), + account_id="acct-1", + buys=world.host.buys["acct-1"], + offerings=OFFERINGS, + account_timezone="America/Los_Angeles", + now=at("2026-10-08T00:00:00Z"), + authorizations=world.grants["acct-1"], + coverage=coverage, + status_retention_days=world.retention_days, + ) + assert buyer == seller + # New York days through Oct 3 (E = Oct 4 04:00Z), then Los Angeles days + # from 07:00Z; 04:00Z to 07:00Z on Oct 4 is owed under neither. + assert [(p.period_start, p.source_timezone) for p in seller] == [ + ("2026-10-01T04:00:00Z", NY), + ("2026-10-02T04:00:00Z", NY), + ("2026-10-03T04:00:00Z", NY), + ("2026-10-04T07:00:00Z", "America/Los_Angeles"), + ("2026-10-05T07:00:00Z", "America/Los_Angeles"), + ("2026-10-06T07:00:00Z", "America/Los_Angeles"), + ] + + +@pytest.mark.asyncio +async def test_gaps_are_per_offering_when_boundaries_differ() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [ + buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-08T04:00:00Z", offering_ids=("pkg",)), + buy("mb-2", "2026-10-01T04:00:00Z", "2026-10-08T04:00:00Z", offering_ids=("pkg-hourly",)), + ] + await world.sync( + subscription(), + subscription( + "pacing", + selector={ + "feed_purpose": "analytics", + "period_duration": "PT1H", + "required_finality": "official", + "dimensions": ["package"], + }, + scope={"account_ids": ["acct-1"]}, + ), + ) + await world.reconcile("2026-10-01T00:00:00Z") + world.revoke("acct-1", "2026-10-03T10:30:00Z") + await world.reconcile("2026-10-03T10:30:00Z") + world.grant("acct-1", "2026-10-04T15:30:00Z") + await world.reconcile("2026-10-04T15:30:00Z") + daily = await world.coverage("acct-1", "analytics") + hourly = await world.coverage("acct-1", "pacing") + # Each subscription resolves one offering, so each gap uses its boundaries. + assert [(g.from_, g.until, g.offering_id) for g in daily.gaps] == [ + (at("2026-10-03T04:00:00Z"), at("2026-10-05T04:00:00Z"), None) + ] + assert [(g.from_, g.until, g.offering_id) for g in hourly.gaps] == [ + (at("2026-10-03T10:00:00Z"), at("2026-10-04T16:00:00Z"), None) + ] + await world.assert_conformant("analytics", "acct-1", "2026-10-07T00:00:00Z", daily) + await world.assert_conformant("pacing", "acct-1", "2026-10-05T00:00:00Z", hourly) + + +@pytest.mark.asyncio +async def test_gap_entries_name_offerings_whose_boundaries_differ() -> None: + # match all resolves a New York day and a UTC day, whose boundaries differ. + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [ + buy( + "mb-1", + "2026-10-01T04:00:00Z", + "2026-10-08T04:00:00Z", + offering_ids=("pkg", "pkg-region-utc"), + ), + ] + await world.sync( + subscription( + selector={ + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "dimensions": ["package"], + "match": "all", + } + ) + ) + await world.reconcile("2026-10-01T00:00:00Z") + world.revoke("acct-1", "2026-10-03T10:30:00Z") + await world.reconcile("2026-10-03T10:30:00Z") + world.grant("acct-1", "2026-10-04T15:30:00Z") + await world.reconcile("2026-10-04T15:30:00Z") + coverage = await world.coverage("acct-1", "analytics") + assert [(g.offering_id, g.from_, g.until) for g in coverage.gaps] == [ + ("pkg", at("2026-10-03T04:00:00Z"), at("2026-10-05T04:00:00Z")), + ("pkg-region-utc", at("2026-10-03T00:00:00Z"), at("2026-10-05T00:00:00Z")), + ] + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-07T05:00:00Z", coverage) + assert {p.offering_id for p in owed} == {"pkg", "pkg-region-utc"} + + +@pytest.mark.asyncio +async def test_receipt_roster_follows_active_receipt_subscriptions() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + await world.principals.sync_principal( + IDENTITY, + { + "idempotency_key": "engine-roster-key-0001", + "configuration": { + "reporting_destinations": [ + { + "pattern": "file_transfer", + "destination_id": "archive", + "active": True, + "provider": {"domain": "object-store.example"}, + "transport": "s3", + "location": "s3://buyer-reporting/adcp/", + "accepted_formats": ["parquet"], + "accepted_verification_profiles": ["manifest_checksums"], + } + ], + "reporting_subscriptions": [ + subscription( + "billing", + selector={ + "feed_purpose": "billing", + "period_duration": "P1D", + "required_finality": "official", + "dimensions": ["package"], + }, + reconciliation_mode="consumer_receipt", + destination={ + "destination_id": "archive", + "path_template": "billing/{account_id}", + }, + ), + subscription(), + ], + }, + }, + ) + await world.reconcile("2026-10-01T00:00:00Z") + assert await world.engine.receipt_roster(IDENTITY, "acct-1") == ["billing"] + assert await world.engine.receipt_roster(IDENTITY, "acct-unknown") == [] + + +@pytest.mark.asyncio +async def test_beta_1_callers_get_account_wide_entries_without_new_fields() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [ + buy( + "mb-1", + "2026-10-01T04:00:00Z", + "2026-10-08T04:00:00Z", + offering_ids=("pkg", "pkg-region-utc"), + ), + ] + await world.sync( + subscription( + selector={ + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "dimensions": ["package"], + "match": "all", + } + ) + ) + await world.reconcile("2026-10-01T00:00:00Z") + world.revoke("acct-1", "2026-10-03T10:30:00Z") + await world.reconcile("2026-10-03T10:30:00Z") + world.grant("acct-1", "2026-10-04T15:30:00Z") + await world.reconcile("2026-10-04T15:30:00Z") + (entry,) = await world.engine.subscription_coverage(IDENTITY, "acct-1") + assert [(g.offering_id, g.from_, g.until) for g in entry.gaps] == [ + (None, at("2026-10-03T00:00:00Z"), at("2026-10-05T04:00:00Z")) + ] + # The served response validates against the pinned 3.3.0-beta.1 schema. + from adcp.validation import schema_loader + + validator = schema_loader.get_named_validator( + "core/reporting-subscription-coverage.json", version="3.3.0-beta.1" + ) + payload = entry.model_dump(mode="json", exclude_none=True, by_alias=True) + assert [error.message for error in validator.iter_errors(payload)] == [] + + +# -- expert review fixes ----------------------------------------------------------------- + +DEST_A = { + "pattern": "file_transfer", + "destination_id": "dest-a", + "active": True, + "provider": {"domain": "object-store.example"}, + "transport": "s3", + "location": "s3://buyer-a/adcp/", + "accepted_formats": ["parquet"], + "accepted_verification_profiles": ["manifest_checksums"], +} +DEST_B = {**DEST_A, "destination_id": "dest-b", "location": "s3://buyer-b/adcp/"} +MANAGED = offering("pkg-file", ("package",), method_pattern="file_transfer", method_transport="s3") + + +async def _sync_raw(world: World, configuration: dict) -> None: + world._key += 1 + response = await world.principals.sync_principal( + IDENTITY, + {"idempotency_key": f"engine-raw-key-{world._key:06d}", "configuration": configuration}, + ) + assert response.result.kind == "applied", response + + +@pytest.mark.asyncio +async def test_removing_a_prior_versions_destination_keeps_its_coverage() -> None: + world = World() + world.engine._offerings = (*world.engine._offerings, MANAGED) + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [ + buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-06T04:00:00Z", offering_ids=("pkg-file",)) + ] + v1 = subscription(destination={"destination_id": "dest-a", "path_template": "x/{account_id}"}) + await _sync_raw( + world, {"reporting_destinations": [DEST_A, DEST_B], "reporting_subscriptions": [v1]} + ) + await world.reconcile("2026-10-01T00:00:00Z") + world.clock.set("2026-10-03T12:00:00Z") + v2 = subscription( + subscription_version=2, + destination={"destination_id": "dest-b", "path_template": "y/{account_id}"}, + ) + await _sync_raw(world, {"reporting_subscriptions": [v2]}) + await _sync_raw(world, {"reporting_destinations": [DEST_B]}) + owed = await world.seller_owes("analytics", "acct-1", "2026-10-07T00:00:00Z") + assert [(p.period_start, p.delivery_config_version) for p in owed] == [ + ("2026-10-01T04:00:00Z", 1), + ("2026-10-02T04:00:00Z", 1), + ("2026-10-03T04:00:00Z", 1), + ("2026-10-04T04:00:00Z", 2), + ("2026-10-05T04:00:00Z", 2), + ] + + +@pytest.mark.asyncio +async def test_a_version_bump_while_inactive_stays_computable() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-08T04:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.clock.set("2026-10-03T12:00:00Z") + await world.sync(subscription(active=False)) + world.clock.set("2026-10-05T00:00:00Z") + await world.sync(subscription(subscription_version=2, active=False)) + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-08T00:00:00Z") + assert [p.period_start for p in owed] == [ + "2026-10-01T04:00:00Z", + "2026-10-02T04:00:00Z", + "2026-10-03T04:00:00Z", + ] + + +@pytest.mark.asyncio +async def test_losing_authority_never_closes_an_override() -> None: + world = World(support=ReportingSubscriptionSupport(backfill_on_regrant=True)) + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-12T04:00:00Z")] + world.host.configs["acct-1"] = [AccountDeliveryConfig("acct-pacing", "analytics")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.revoke("acct-1", "2026-10-05T10:00:00Z") + await world.reconcile("2026-10-05T10:00:00Z") + world.grant("acct-1", "2026-10-08T10:00:00Z") + await world.reconcile("2026-10-08T10:00:00Z") + coverage = await world.coverage("acct-1", "analytics") + assert [(e.reason, e.until) for e in coverage.excluded] == [("overridden", None)] + # The account config owns the account throughout: the subscription never + # owes a second feed, not even for the backfilled revoked interval. + assert ( + await world.assert_conformant("analytics", "acct-1", "2026-10-11T00:00:00Z", coverage) == [] + ) + + +@pytest.mark.asyncio +async def test_exceptions_use_the_earliest_cutover_across_calendars() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [ + buy( + "mb-1", + "2026-10-01T04:00:00Z", + "2026-10-09T04:00:00Z", + offering_ids=("pkg", "pkg-region-utc"), + ), + ] + await world.sync( + subscription( + selector={ + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "dimensions": ["package"], + "match": "all", + } + ) + ) + await world.reconcile("2026-10-01T00:00:00Z") + world.host.configs["acct-1"] = [AccountDeliveryConfig("acme", "analytics")] + await world.reconcile("2026-10-05T12:00:00Z") + coverage = await world.coverage("acct-1", "analytics") + # UTC day boundary 10-06T00Z comes before the New York one (10-06T04Z). + assert coverage.excluded[0].from_ == at("2026-10-06T00:00:00Z") + owed = await world.assert_conformant("analytics", "acct-1", "2026-10-09T05:00:00Z", coverage) + assert max(at(p.period_start) for p in owed) < at("2026-10-06T00:00:00Z") + + +@pytest.mark.asyncio +async def test_an_override_opens_at_its_own_effective_time() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-09T04:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.host.configs["acct-1"] = [ + AccountDeliveryConfig("acme", "analytics", active_from=at("2026-10-03T10:00:00Z")) + ] + # Reconciled two days late: the exception still opens at the config's cutover. + await world.reconcile("2026-10-05T10:00:00Z") + coverage = await world.coverage("acct-1", "analytics") + assert coverage.excluded[0].from_ == at("2026-10-04T04:00:00Z") + + +@pytest.mark.asyncio +async def test_an_account_without_a_timezone_is_never_given_utc() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-05T04:00:00Z")] + await world.sync(subscription()) + # The host reports the grant without a zone: account-timezone offerings + # cannot be scheduled, and the failure stays with this subscription. + del world.host.accounts["acct-1"] + await world.engine.authorization_changed( + IDENTITY, "acct-1", authorized=True, at=at("2026-09-01T00:00:00Z") + ) + await world.engine._evaluate(IDENTITY, at("2026-10-01T00:00:00Z")) + assert ("https://agency.example/agent", "analytics") in world.engine.evaluation_errors + with pytest.raises(ValueError, match="account timezone is required"): + await world.seller_owes("analytics", "acct-1", "2026-10-06T00:00:00Z") + # Once the host says, the zone is set without a fake UTC -> NY cutover. + world.host.accounts["acct-1"] = AuthorizedAccount( + "acct-1", timezone=NY, authorized_at=at("2026-09-01T00:00:00Z") + ) + await world.reconcile("2026-10-01T00:00:00Z") + coverage = await world.coverage("acct-1", "analytics") + assert coverage.timezone_cutovers is None + assert world.engine.evaluation_errors == {} + owed = await world.seller_owes("analytics", "acct-1", "2026-10-06T00:00:00Z") + assert [p.source_timezone for p in owed] == [NY] * 4 + + +@pytest.mark.asyncio +async def test_timezone_change_then_new_version_in_one_period_matches_the_wire() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z", timezone="America/Los_Angeles") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T07:00:00Z", "2026-10-07T04:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.host.accounts["acct-1"] = AuthorizedAccount( + "acct-1", timezone=NY, authorized_at=at("2026-09-01T00:00:00Z") + ) + await world.reconcile("2026-10-03T12:00:00Z") + world.clock.set("2026-10-03T13:00:00Z") + await world.sync(subscription(subscription_version=2)) + coverage = await world.coverage("acct-1", "analytics") + seller = await world.seller_owes("analytics", "acct-1", "2026-10-07T05:00:00Z") + buyer = expected_subscription_periods( + subscription=await world.state("analytics"), + account_id="acct-1", + buys=world.host.buys["acct-1"], + offerings=OFFERINGS, + account_timezone=NY, + now=at("2026-10-07T05:00:00Z"), + authorizations=world.grants["acct-1"], + coverage=coverage, + status_retention_days=world.retention_days, + ) + assert buyer == seller + spans = [(at(p.period_start), at(p.period_end)) for p in seller] + assert all(end <= start for (_, end), (start, _) in zip(spans, spans[1:])) + + +@pytest.mark.asyncio +async def test_unauthorized_lists_only_accounts_the_subscription_served() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.configs["acct-1"] = [AccountDeliveryConfig("acme", "analytics")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.revoke("acct-1", "2026-10-03T00:00:00Z") + await world.reconcile("2026-10-03T00:00:00Z") + assert (await world.state("analytics")).accounts_unauthorized is None + + +@pytest.mark.asyncio +async def test_delivery_facts_are_asked_once_per_loss() -> None: + world = World() + calls: list[str] = [] + + async def is_period_delivered(principal, account_id, period, *, by): + calls.append(period.period_start) + return False + + world.host.is_period_delivered = is_period_delivered # type: ignore[attr-defined] + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-12T04:00:00Z")] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + world.revoke("acct-1", "2026-10-04T08:00:00Z") + await world.reconcile("2026-10-04T08:00:00Z") + first = len(calls) + await world.reconcile("2026-10-05T08:00:00Z") + await world.reconcile("2026-10-06T08:00:00Z") + assert first == 3 and len(calls) == first + + +def test_engine_refuses_offerings_it_cannot_schedule() -> None: + unresolved = offering( + "src", ("package",), alignment="source_timezone", period_timezone_policy="account_resolved" + ) + seven_hours = offering("local-7h", ("package",), period_duration="PT7H") + for bad in (unresolved, seven_hours): + with pytest.raises(ValueError): + ReportingSubscriptionEngine(principals=PrincipalService(), host=Host(), offerings=[bad])