From fe87948db5dda59288b621cd8a109af703d7380e Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 10:00:40 +0000 Subject: [PATCH 1/4] feat(reporting): add buyer client helpers and a tracker for reporting subscriptions Buyer side of AdCP 3.3 reporting subscriptions: - sync_reporting_subscriptions() writes the sync_principal section, optionally with reporting_destinations, and returns the readback. read_reporting_subscriptions() reads it back through get_principal. load_subscription_coverage() reads subscription_coverage[] from the get_reporting_status summary view. Seller rejections raise ReportingSubscriptionClientError carrying the error code. - ReportingSubscriptionTracker keeps the buyer's own records: - every registered version's configuration; - grant history from list_accounts and accounts_unauthorized[]; - timezone changes. It re-reads on a relevant principal.changed notification, and expected_periods() computes what the seller owes from those records alone. Results are ExpectedReportingPeriod values, so they feed evaluate_reporting_ledger and plan_consumer_statuses next to configuration-derived expectations; withheld periods carry no expected_at and are never claimed missing. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reliable-reporting-service.md | 12 + src/adcp/reporting/subscriptions/__init__.py | 21 + src/adcp/reporting/subscriptions/consumer.py | 412 ++++++++++++++++++ tests/test_reporting_subscription_consumer.py | 244 +++++++++++ 4 files changed, 689 insertions(+) create mode 100644 src/adcp/reporting/subscriptions/consumer.py create mode 100644 tests/test_reporting_subscription_consumer.py diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index 581f2f8ba..68007ec8f 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -334,6 +334,18 @@ and the `subscription_coverage[]` exceptions and gaps. It returns they feed `evaluate_reporting_ledger` and the consumer-status planner as they are. +`ReportingSubscriptionTracker` keeps those records for you. Register through +`tracker.sync(...)` so every version's configuration is kept (the readback +keeps only the current one), record what `list_accounts` shows with +`tracker.observe_account(...)`, and pass each `principal.changed` payload to +`tracker.handle_notification(...)`; it re-reads `get_principal` and folds +`accounts_unauthorized[]` into the grant history. +`tracker.expected_periods(subscription_id, account_id, buys=..., now=..., +coverage=...)` then returns what the seller owes. A withheld period carries no +`expected_at`, so it is never claimed missing while authority is absent. +`sync_reporting_subscriptions`, `read_reporting_subscriptions` and +`load_subscription_coverage` are the underlying client helpers. + **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. diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index bd2bdc82c..4cfa39e13 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -15,6 +15,9 @@ * :func:`apply_reporting_subscriptions` -- validation and versioning of the ``sync_principal`` section, used by ``PrincipalService`` when it is given ``reporting_subscriptions=`` options. +* :class:`ReportingSubscriptionTracker` and :func:`sync_reporting_subscriptions` + -- the buyer's client side: register, read back, refresh on + ``principal.changed``, and compute expected periods from its own records. * :class:`ReportingSubscriptionEngine` -- the seller runtime: host hooks, lost authority, exceptions, ``subscription_coverage[]`` and owed periods. """ @@ -30,6 +33,16 @@ subscription_capabilities, with_subscription_dimensions, ) +from adcp.reporting.subscriptions.consumer import ( + REPORTING_SUBSCRIPTION_CHANGED, + ReportingSubscriptionClientError, + ReportingSubscriptionReadback, + ReportingSubscriptionTracker, + is_subscription_change, + load_subscription_coverage, + read_reporting_subscriptions, + sync_reporting_subscriptions, +) from adcp.reporting.subscriptions.coverage import ( AccountTimezoneChange, ExpectedSubscriptionPeriod, @@ -78,6 +91,14 @@ ) __all__ = [ + "REPORTING_SUBSCRIPTION_CHANGED", + "ReportingSubscriptionClientError", + "ReportingSubscriptionReadback", + "ReportingSubscriptionTracker", + "is_subscription_change", + "load_subscription_coverage", + "read_reporting_subscriptions", + "sync_reporting_subscriptions", "AccountDeliveryConfig", "AuthorizedAccount", "AuthorizedAccountPage", diff --git a/src/adcp/reporting/subscriptions/consumer.py b/src/adcp/reporting/subscriptions/consumer.py new file mode 100644 index 000000000..061f015da --- /dev/null +++ b/src/adcp/reporting/subscriptions/consumer.py @@ -0,0 +1,412 @@ +"""Buyer side of reporting subscriptions: register, read back, and stay current. + +* :func:`sync_reporting_subscriptions` writes the ``sync_principal`` + ``reporting_subscriptions`` section and returns the seller's readback. +* :func:`read_reporting_subscriptions` reads it back through ``get_principal``. +* :func:`load_subscription_coverage` reads ``subscription_coverage[]`` from the + ``get_reporting_status`` summary view for one account. +* :class:`ReportingSubscriptionTracker` keeps the readback current, refreshes + when a ``principal.changed`` notification arrives, remembers every version + the buyer registered (the readback keeps only the current configuration), + tracks lost authority from ``accounts_unauthorized[]``, and computes the + buyer's expected periods with + :func:`~adcp.reporting.subscriptions.expected_subscription_periods`. + +The expected periods are :class:`~adcp.reporting.ExpectedReportingPeriod` +values, so they go straight into +:func:`~adcp.reporting.evaluate_reporting_ledger` (``expected_periods=``) next +to configuration-derived expectations, and from there into +:func:`~adcp.reporting.plan_consumer_statuses`. A withheld period has no +``expected_at`` and is never claimed missing while authority is absent. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Protocol +from uuid import uuid4 + +from adcp.reporting.subscriptions._clock import utc_instant +from adcp.reporting.subscriptions.coverage import ( + AccountTimezoneChange, + ExpectedSubscriptionPeriod, + SubscriptionAuthorization, + SubscriptionBuy, + expected_subscription_periods, +) +from adcp.reporting.subscriptions.resolution import SubscriptionOffering +from adcp.types import ( + GetPrincipalRequest, + GetPrincipalResponse, + GetReportingStatusRequest, + GetReportingStatusResponse, + SyncPrincipalRequest, + SyncPrincipalResponse, +) +from adcp.types.core import TaskResult +from adcp.types.domains.core.reporting_subscription_coverage import ( + ReportingSubscriptionCoverage, +) +from adcp.types.domains.core.reporting_subscription_state import ReportingSubscriptionState + +#: ``principal.changed`` reason that means subscription state changed. +REPORTING_SUBSCRIPTION_CHANGED = "reporting_subscription_changed" + + +class ReportingSubscriptionClientError(RuntimeError): + """The seller refused or failed a subscription read or write.""" + + def __init__(self, message: str, *, code: str | None = None) -> None: + super().__init__(message) + self.code = code + + +class PrincipalClient(Protocol): + async def get_principal( + self, request: GetPrincipalRequest + ) -> TaskResult[GetPrincipalResponse]: ... + + async def sync_principal( + self, request: SyncPrincipalRequest + ) -> TaskResult[SyncPrincipalResponse]: ... + + +class SubscriptionStatusClient(Protocol): + async def get_reporting_status( + self, request: GetReportingStatusRequest + ) -> TaskResult[GetReportingStatusResponse]: ... + + +@dataclass(frozen=True) +class ReportingSubscriptionReadback: + """The caller's subscriptions as the seller reports them.""" + + configuration_version: str | None + subscriptions: tuple[ReportingSubscriptionState, ...] = () + + def get(self, subscription_id: str) -> ReportingSubscriptionState | None: + return next( + (item for item in self.subscriptions if item.subscription_id == subscription_id), + None, + ) + + +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 _readback(result: Any, *, task: str) -> ReportingSubscriptionReadback: + response = getattr(result, "data", None) + if not getattr(result, "success", False) or response is None: + raise ReportingSubscriptionClientError( + f"{task} failed: {getattr(result, 'error', None) or 'no response'}" + ) + outcome = response.result + kind = str(getattr(outcome, "kind", "")) + if kind == "failed": + errors = list(getattr(outcome, "errors", None) or []) + first = errors[0] if errors else None + raise ReportingSubscriptionClientError( + f"{task} was rejected: {getattr(first, 'message', 'unknown error')}", + code=None if first is None else str(getattr(first, "code", "")), + ) + configuration = getattr(outcome, "configuration", None) + states = tuple(getattr(configuration, "reporting_subscriptions", None) or ()) + return ReportingSubscriptionReadback( + configuration_version=getattr(outcome, "configuration_version", None), + subscriptions=states, + ) + + +async def sync_reporting_subscriptions( + client: PrincipalClient, + subscriptions: Sequence[Any], + *, + reporting_destinations: Sequence[Any] | None = None, + idempotency_key: str | None = None, + expected_configuration_version: str | None = None, + dry_run: bool = False, +) -> ReportingSubscriptionReadback: + """Write the complete desired subscription set and return the readback. + + The section is complete desired state: a subscription missing from + ``subscriptions`` is deactivated. Pass ``reporting_destinations`` to + replace that section in the same request, so a new subscription can name + a destination registered alongside it. A ``dry_run`` validates without + applying and returns an empty readback. + """ + configuration: dict[str, Any] = { + "reporting_subscriptions": [_dump(item) for item in subscriptions] + } + if reporting_destinations is not None: + configuration["reporting_destinations"] = [_dump(item) for item in reporting_destinations] + payload: dict[str, Any] = { + "idempotency_key": idempotency_key or f"rsub-{uuid4()}", + "configuration": configuration, + } + if expected_configuration_version is not None: + payload["expected_configuration_version"] = expected_configuration_version + if dry_run: + payload["dry_run"] = True + result = await client.sync_principal(SyncPrincipalRequest.model_validate(payload)) + return _readback(result, task="sync_principal") + + +async def read_reporting_subscriptions(client: PrincipalClient) -> ReportingSubscriptionReadback: + """Read the caller's subscriptions back through ``get_principal``.""" + result = await client.get_principal(GetPrincipalRequest.model_validate({})) + response = getattr(result, "data", None) + if response is not None and str(getattr(response.result, "kind", "")) in { + "unconfigured", + "recognized", + }: + return ReportingSubscriptionReadback(configuration_version=None) + return _readback(result, task="get_principal") + + +async def load_subscription_coverage( + client: SubscriptionStatusClient, *, account: Mapping[str, Any] | Any +) -> tuple[ReportingSubscriptionCoverage, ...]: + """``subscription_coverage[]`` for one account from the summary view. + + An account the caller has lost authority for is unknown to + ``get_reporting_status``; read ``accounts_unauthorized[]`` from + :func:`read_reporting_subscriptions` instead. + """ + request = GetReportingStatusRequest.model_validate( + {"account": _dump(account), "view": "summary"} + ) + result = await client.get_reporting_status(request) + response = getattr(result, "data", None) + if not getattr(result, "success", False) or response is None: + raise ReportingSubscriptionClientError( + f"get_reporting_status failed: {getattr(result, 'error', None) or 'no response'}" + ) + return tuple(getattr(response, "subscription_coverage", None) or ()) + + +def is_subscription_change(notification: Mapping[str, Any] | Any) -> bool: + """Whether a ``principal.changed`` payload calls for re-reading subscriptions. + + Receivers re-read rather than infer state from the reason; ``other`` and + destination changes can move a subscription to ``action_required`` too. + """ + payload = _dump(notification) + if payload.get("notification_type") not in {None, "principal.changed"}: + return False + return str(payload.get("reason")) in { + REPORTING_SUBSCRIPTION_CHANGED, + "destination_state_changed", + "other", + } + + +@dataclass +class _AccountRecord: + timezone: str + operator: str | None = None + initial_timezone: str = "" + grants: list[SubscriptionAuthorization] = field(default_factory=list) + timezone_changes: list[AccountTimezoneChange] = field(default_factory=list) + + +class ReportingSubscriptionTracker: + """A buyer's own records for its subscriptions with one seller. + + Register through :meth:`sync` so every version's configuration is kept. + Record what :func:`list_accounts` shows with :meth:`observe_account`. + Call :meth:`handle_notification` with each ``principal.changed`` payload + (or :meth:`refresh` on a schedule). :meth:`expected_periods` then computes + what the seller owes from these records alone. + """ + + def __init__( + self, + client: PrincipalClient, + *, + offerings: Sequence[SubscriptionOffering], + backfill_on_regrant: bool = False, + status_retention_days: int | None = None, + destinations: Mapping[str, Any] | None = None, + ) -> None: + self._client = client + self._offerings = tuple(offerings) + self._backfill = backfill_on_regrant + self._retention = status_retention_days + self._destinations = dict(destinations or {}) + self._sent: dict[str, dict[int, dict[str, Any]]] = {} + self._accounts: dict[str, _AccountRecord] = {} + self._readback = ReportingSubscriptionReadback(configuration_version=None) + + @classmethod + def from_capabilities( + cls, + client: PrincipalClient, + capabilities: Mapping[str, Any] | Any, + *, + offerings: Sequence[SubscriptionOffering], + destinations: Mapping[str, Any] | None = None, + ) -> ReportingSubscriptionTracker: + """Read ``backfill_on_regrant`` and ``status_retention_days`` from capabilities.""" + payload = _dump(capabilities) + block = (payload.get("media_buy") or {}).get("reporting_delivery") or payload + subscriptions = block.get("subscriptions") or {} + return cls( + client, + offerings=offerings, + backfill_on_regrant=bool(subscriptions.get("backfill_on_regrant", False)), + status_retention_days=block.get("status_retention_days"), + destinations=destinations, + ) + + @property + def readback(self) -> ReportingSubscriptionReadback: + return self._readback + + async def sync( + self, + subscriptions: Sequence[Any], + *, + reporting_destinations: Sequence[Any] | None = None, + idempotency_key: str | None = None, + expected_configuration_version: str | None = None, + ) -> ReportingSubscriptionReadback: + """Register the desired set, remembering each version's configuration.""" + payloads = [_dump(item) for item in subscriptions] + readback = await sync_reporting_subscriptions( + self._client, + payloads, + reporting_destinations=reporting_destinations, + idempotency_key=idempotency_key, + expected_configuration_version=expected_configuration_version, + ) + for item in payloads: + versions = self._sent.setdefault(str(item["subscription_id"]), {}) + versions.setdefault(int(item["subscription_version"]), item) + if reporting_destinations is not None: + for destination in reporting_destinations: + payload = _dump(destination) + self._destinations[str(payload["destination_id"])] = payload + self._readback = readback + return readback + + async def refresh(self) -> ReportingSubscriptionReadback: + """Re-read the readback and fold ``accounts_unauthorized[]`` into grant history.""" + self._readback = await read_reporting_subscriptions(self._client) + for state in self._readback.subscriptions: + for entry in state.accounts_unauthorized or (): + self._record_loss(entry.account_id, utc_instant(entry.unauthorized_at)) + return self._readback + + async def handle_notification(self, notification: Mapping[str, Any] | Any) -> bool: + """Refresh on a relevant ``principal.changed``; returns whether it refreshed.""" + if not is_subscription_change(notification): + return False + await self.refresh() + return True + + def observe_account( + self, + account_id: str, + *, + authorized_at: datetime | None, + timezone: str = "UTC", + operator: str | None = None, + observed_at: datetime | None = None, + ) -> None: + """Record what ``list_accounts`` shows for an account the caller can read. + + A new ``authorized_at`` after a recorded loss is a re-grant. A changed + ``timezone`` is recorded as a timezone change at ``observed_at``. + """ + record = self._accounts.get(account_id) + if record is None: + record = self._accounts[account_id] = _AccountRecord( + timezone, operator, initial_timezone=timezone + ) + elif timezone != record.timezone: + if observed_at is None: + raise ValueError("a timezone change needs observed_at") + record.timezone_changes.append( + AccountTimezoneChange(utc_instant(observed_at), timezone) + ) + record.timezone = timezone + if operator is not None: + record.operator = operator + if authorized_at is None: + return + granted = utc_instant(authorized_at, field="authorized_at") + if not record.grants: + record.grants.append(SubscriptionAuthorization(granted)) + return + last = record.grants[-1] + if last.unauthorized_at is not None and granted > last.unauthorized_at: + record.grants.append(SubscriptionAuthorization(granted)) + + def _record_loss(self, account_id: str, unauthorized_at: datetime) -> None: + record = self._accounts.get(account_id) + if record is None or not record.grants: + return + last = record.grants[-1] + if last.unauthorized_at is None and unauthorized_at > last.authorized_at: + record.grants[-1] = SubscriptionAuthorization(last.authorized_at, unauthorized_at) + + def expected_periods( + self, + subscription_id: str, + account_id: str, + *, + buys: Sequence[SubscriptionBuy], + now: datetime, + coverage: ReportingSubscriptionCoverage | None = None, + include_unresolved: bool = False, + ) -> list[ExpectedSubscriptionPeriod]: + """What the seller owes this account under the subscription, from buyer records. + + ``coverage`` is the subscription's ``subscription_coverage[]`` entry + for the account, read while the caller is authorized. + """ + state = self._readback.get(subscription_id) + if state is None: + raise KeyError(subscription_id) + record = self._accounts.get(account_id) + if record is None: + raise KeyError(account_id) + return expected_subscription_periods( + subscription=state, + version_configurations=self._sent.get(subscription_id), + account_id=account_id, + buys=buys, + offerings=self._offerings, + account_timezone=record.initial_timezone, + now=now, + authorizations=list(record.grants), + exceptions=() if coverage is None else coverage.excluded, + gaps=() if coverage is None else (coverage.gaps or ()), + backfill_on_regrant=self._backfill, + status_retention_days=self._retention, + destinations=self._destinations, + account_operator=record.operator, + timezone_changes=list(record.timezone_changes), + include_unresolved=include_unresolved, + ) + + +__all__ = [ + "REPORTING_SUBSCRIPTION_CHANGED", + "PrincipalClient", + "ReportingSubscriptionClientError", + "ReportingSubscriptionReadback", + "ReportingSubscriptionTracker", + "SubscriptionStatusClient", + "is_subscription_change", + "load_subscription_coverage", + "read_reporting_subscriptions", + "sync_reporting_subscriptions", +] diff --git a/tests/test_reporting_subscription_consumer.py b/tests/test_reporting_subscription_consumer.py new file mode 100644 index 000000000..d2e646e47 --- /dev/null +++ b/tests/test_reporting_subscription_consumer.py @@ -0,0 +1,244 @@ +"""Buyer client helpers and tracker for reporting subscriptions (AdCP 3.3). + +A real in-process seller (``PrincipalService`` plus ``ReportingSubscriptionEngine``) +sits behind a thin client adapter, so the buyer sees only wire responses. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta + +import pytest + +from adcp.reporting import plan_consumer_statuses +from adcp.reporting.subscriptions import ( + AuthorizedAccount, + AuthorizedAccountPage, + ReportingSubscriptionClientError, + ReportingSubscriptionEngine, + ReportingSubscriptionTracker, + SubscriptionBuy, + SubscriptionOffering, + is_subscription_change, + load_subscription_coverage, + read_reporting_subscriptions, + sync_reporting_subscriptions, +) +from adcp.server import PrincipalChange, PrincipalIdentity, PrincipalService +from adcp.types import GetReportingStatusResponse +from adcp.types.core import TaskResult, TaskStatus + +IDENTITY = PrincipalIdentity("https://agency.example/agent", "buyer_agent") +NY = "America/New_York" + + +def at(text: str) -> datetime: + return datetime.fromisoformat(text.replace("Z", "+00:00")) + + +DAILY = SubscriptionOffering( + offering_id="pkg", + report_definition_id="def-pkg", + reporting_profile="delivery", + feed_purpose="analytics", + period_duration="P1D", + alignment="account_timezone", + delivery_sla="PT6H", + supported_finality=frozenset({"official"}), + reconciliation_mode="delivery_only", + dimensions=frozenset({"package"}), +) +SUBSCRIPTION = { + "subscription_id": "analytics", + "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", +} +BUY = SubscriptionBuy( + media_buy_id="mb-1", + confirmed_at=at("2026-10-01T00:00:00Z"), + start_time=at("2026-10-01T04:00:00Z"), + end_time=at("2026-10-12T04:00:00Z"), + reporting_delivery_offering_ids=("pkg",), +) + + +class Seller: + def __init__(self) -> None: + self.now = at("2026-10-01T00:00:00Z") + self.accounts: dict[str, AuthorizedAccount] = {} + self.notifications: list[dict] = [] + + async def emit(change: PrincipalChange) -> None: + self.notifications.append( + {"notification_type": "principal.changed", "reason": change.reason} + ) + + self.principals = PrincipalService(emit_change=emit) + self.engine = ReportingSubscriptionEngine( + principals=self.principals, host=self, offerings=[DAILY], clock=lambda: self.now + ) + self.principals.set_reporting_subscriptions(self.engine.principal_options()) + + async def list_authorized_accounts(self, principal, *, cursor): + return AuthorizedAccountPage(tuple(self.accounts.values())) + + async def account_buys(self, principal, account_id): + return [BUY] if account_id in {"acct-1"} else [] + + async def reconcile(self, when: str) -> None: + self.now = at(when) + await self.engine.reconcile(IDENTITY) + + +class Client: + """What an ADCPClient returns, backed by the in-process seller.""" + + def __init__(self, seller: Seller) -> None: + self.seller = seller + + async def sync_principal(self, request): + data = await self.seller.principals.sync_principal(IDENTITY, request) + return TaskResult(status=TaskStatus.COMPLETED, data=data, success=True) + + async def get_principal(self, request): + data = await self.seller.principals.get_principal(IDENTITY, request) + return TaskResult(status=TaskStatus.COMPLETED, data=data, success=True) + + async def get_reporting_status(self, request): + account_id = request.account.account_id + coverage = await self.seller.engine.subscription_coverage(IDENTITY, account_id) + data = GetReportingStatusResponse.model_construct(subscription_coverage=coverage or None) + return TaskResult(status=TaskStatus.COMPLETED, data=data, success=True) + + +@pytest.mark.asyncio +async def test_sync_and_read_back() -> None: + seller = Seller() + client = Client(seller) + assert (await read_reporting_subscriptions(client)).subscriptions == () + readback = await sync_reporting_subscriptions(client, [SUBSCRIPTION]) + state = readback.get("analytics") + assert state is not None and state.state == "active" + assert readback.configuration_version + again = await read_reporting_subscriptions(client) + assert again.get("analytics").accepted_at == at("2026-10-01T00:00:00Z") + dry = await sync_reporting_subscriptions( + client, [{**SUBSCRIPTION, "subscription_version": 2}], dry_run=True + ) + assert dry.subscriptions == () + + +@pytest.mark.asyncio +async def test_rejection_surfaces_the_seller_code() -> None: + client = Client(Seller()) + with pytest.raises(ReportingSubscriptionClientError) as raised: + await sync_reporting_subscriptions(client, [{**SUBSCRIPTION, "scope": {}}]) + assert raised.value.code == "INVALID_REQUEST" + + +def test_notification_filter() -> None: + assert is_subscription_change( + {"notification_type": "principal.changed", "reason": "reporting_subscription_changed"} + ) + assert is_subscription_change({"reason": "destination_state_changed"}) + assert not is_subscription_change({"reason": "declarations_intersection_changed"}) + assert not is_subscription_change( + {"notification_type": "account.status_changed", "reason": "other"} + ) + + +@pytest.mark.asyncio +async def test_tracker_follows_revocation_and_regrant_and_matches_the_seller() -> None: + seller = Seller() + client = Client(seller) + tracker = ReportingSubscriptionTracker.from_capabilities( + client, + { + "media_buy": { + "reporting_delivery": { + "status_retention_days": 400, + "subscriptions": {"supported": True}, + } + } + }, + offerings=[DAILY], + ) + seller.accounts["acct-1"] = AuthorizedAccount( + "acct-1", timezone=NY, authorized_at=at("2026-09-01T00:00:00Z") + ) + tracker.observe_account("acct-1", authorized_at=at("2026-09-01T00:00:00Z"), timezone=NY) + await tracker.sync([SUBSCRIPTION]) + await seller.reconcile("2026-10-01T00:00:00Z") + + # Authority ends; the seller says so through principal.changed. + del seller.accounts["acct-1"] + await seller.reconcile("2026-10-04T08:00:00Z") + notification = seller.notifications[-1] + assert notification["reason"] == "reporting_subscription_changed" + assert await tracker.handle_notification(notification) + unauthorized = tracker.readback.get("analytics").accounts_unauthorized + assert [entry.account_id for entry in unauthorized] == ["acct-1"] + assert await load_subscription_coverage(client, account={"account_id": "acct-1"}) == () + + now = at("2026-10-05T00:00:00Z") + buyer = tracker.expected_periods("analytics", "acct-1", buys=[BUY], now=now) + seller_owes = await seller.engine.owed_periods(IDENTITY, "analytics", "acct-1", now=now) + assert buyer == seller_owes + assert [p.withheld for p in buyer] == [False, False, True] + + # Re-grant: list_accounts shows the new authorized_at. + regrant = at("2026-10-06T15:00:00Z") + seller.accounts["acct-1"] = AuthorizedAccount("acct-1", timezone=NY, authorized_at=regrant) + await seller.reconcile("2026-10-06T15:00:00Z") + await tracker.handle_notification(seller.notifications[-1]) + tracker.observe_account("acct-1", authorized_at=regrant, timezone=NY) + (coverage,) = await load_subscription_coverage(client, account={"account_id": "acct-1"}) + assert coverage.gaps + now = at("2026-10-10T00:00:00Z") + buyer = tracker.expected_periods("analytics", "acct-1", buys=[BUY], now=now, coverage=coverage) + assert buyer == await seller.engine.owed_periods(IDENTITY, "analytics", "acct-1", now=now) + + # Subscription expectations feed the consumer-status planner unchanged: + # missing periods past their deadline become obligation_missing, keyed by + # the subscription, and a withheld period is never claimed. + plan = plan_consumer_statuses( + [], + now=now + timedelta(days=1), + automated_recovery_window=timedelta(hours=6), + missing_expected_periods=buyer, + account_id="acct-1", + ) + assert {intent.consumer_status for intent in plan} == {"obligation_missing"} + assert {intent.delivery_config_id for intent in plan} == {"analytics"} + assert len(plan) == len(buyer) + + +@pytest.mark.asyncio +async def test_tracker_keeps_every_registered_version() -> None: + seller = Seller() + client = Client(seller) + tracker = ReportingSubscriptionTracker(client, offerings=[DAILY]) + seller.accounts["acct-1"] = AuthorizedAccount("acct-1", timezone=NY) + tracker.observe_account("acct-1", authorized_at=at("2026-09-01T00:00:00Z"), timezone=NY) + await tracker.sync([SUBSCRIPTION]) + seller.now = at("2026-10-03T12:00:00Z") + narrowed = {**SUBSCRIPTION, "subscription_version": 2, "scope": {"account_ids": ["acct-2"]}} + await tracker.sync([narrowed]) + periods = tracker.expected_periods( + "analytics", "acct-1", buys=[BUY], now=at("2026-10-08T00:00:00Z") + ) + # Version 1 covered acct-1 until version 2 dropped it: Oct 1-3 local days. + assert [p.period_start for p in periods] == [ + "2026-10-01T04:00:00Z", + "2026-10-02T04:00:00Z", + "2026-10-03T04:00:00Z", + ] + assert {p.delivery_config_version for p in periods} == {1} From 104607cee5d7d5ff4f1fea9e33e966b77bed66b8 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 14:27:53 +0000 Subject: [PATCH 2/4] feat(reporting): track re-grants that keep authorized_at, as adcp#8585 settles A re-grant within status_retention_days continues the authorization chain and leaves list_accounts' authorized_at unchanged, so the wire carries no re-grant instant. ReportingSubscriptionTracker.observe_account records the re-grant at observed_at when an account it saw lose authority is readable again with the same authorized_at. A later authorized_at still starts a new chain. The seller's gaps[] remain authoritative for unowed periods. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/consumer.py | 34 +++++++++++---- tests/test_reporting_subscription_consumer.py | 43 +++++++++++++++++-- 2 files changed, 65 insertions(+), 12 deletions(-) diff --git a/src/adcp/reporting/subscriptions/consumer.py b/src/adcp/reporting/subscriptions/consumer.py index 061f015da..afdd80fc9 100644 --- a/src/adcp/reporting/subscriptions/consumer.py +++ b/src/adcp/reporting/subscriptions/consumer.py @@ -211,7 +211,6 @@ def is_subscription_change(notification: Mapping[str, Any] | Any) -> bool: class _AccountRecord: timezone: str operator: str | None = None - initial_timezone: str = "" grants: list[SubscriptionAuthorization] = field(default_factory=list) timezone_changes: list[AccountTimezoneChange] = field(default_factory=list) @@ -322,19 +321,21 @@ def observe_account( ) -> None: """Record what ``list_accounts`` shows for an account the caller can read. - A new ``authorized_at`` after a recorded loss is a re-grant. A changed + After a recorded loss, a later ``authorized_at`` is a re-grant that + started a new chain. An unchanged ``authorized_at`` is a re-grant within + retention (adcp#8585), recorded at ``observed_at``. A changed ``timezone`` is recorded as a timezone change at ``observed_at``. """ record = self._accounts.get(account_id) if record is None: - record = self._accounts[account_id] = _AccountRecord( - timezone, operator, initial_timezone=timezone - ) + record = self._accounts[account_id] = _AccountRecord(timezone, operator) elif timezone != record.timezone: if observed_at is None: raise ValueError("a timezone change needs observed_at") record.timezone_changes.append( - AccountTimezoneChange(utc_instant(observed_at), timezone) + AccountTimezoneChange( + utc_instant(observed_at), timezone, previous_timezone=record.timezone + ) ) record.timezone = timezone if operator is not None: @@ -346,8 +347,25 @@ def observe_account( record.grants.append(SubscriptionAuthorization(granted)) return last = record.grants[-1] - if last.unauthorized_at is not None and granted > last.unauthorized_at: + if last.unauthorized_at is None: + return + if granted > last.unauthorized_at: + # A new chain: the seller reset authorized_at to the re-grant. record.grants.append(SubscriptionAuthorization(granted)) + return + # adcp#8585: a re-grant within status_retention_days continues the + # chain and leaves authorized_at unchanged, so the wire carries no + # re-grant instant. Record the instant this account was observed + # readable again; the seller's gaps[] (when it does not backfill) + # remain authoritative for the periods that are not owed. + if observed_at is None: + raise ValueError( + "the account is readable again with an unchanged authorized_at; " + "pass observed_at so the re-grant can be recorded" + ) + record.grants.append( + SubscriptionAuthorization(max(utc_instant(observed_at), last.unauthorized_at)) + ) def _record_loss(self, account_id: str, unauthorized_at: datetime) -> None: record = self._accounts.get(account_id) @@ -384,7 +402,7 @@ def expected_periods( account_id=account_id, buys=buys, offerings=self._offerings, - account_timezone=record.initial_timezone, + account_timezone=record.timezone, now=now, authorizations=list(record.grants), exceptions=() if coverage is None else coverage.excluded, diff --git a/tests/test_reporting_subscription_consumer.py b/tests/test_reporting_subscription_consumer.py index d2e646e47..7218bf485 100644 --- a/tests/test_reporting_subscription_consumer.py +++ b/tests/test_reporting_subscription_consumer.py @@ -7,6 +7,7 @@ from __future__ import annotations from datetime import datetime, timedelta +from types import SimpleNamespace import pytest @@ -29,6 +30,8 @@ from adcp.types.core import TaskResult, TaskStatus IDENTITY = PrincipalIdentity("https://agency.example/agent", "buyer_agent") +#: The in-process client is 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" @@ -99,22 +102,24 @@ async def reconcile(self, when: str) -> None: class Client: - """What an ADCPClient returns, backed by the in-process seller.""" + """What an ADCPClient returns, backed by the in-process seller (served 3.3.0-beta.2).""" def __init__(self, seller: Seller) -> None: self.seller = seller async def sync_principal(self, request): - data = await self.seller.principals.sync_principal(IDENTITY, request) + data = await self.seller.principals.sync_principal(IDENTITY, request, context=BETA_2) return TaskResult(status=TaskStatus.COMPLETED, data=data, success=True) async def get_principal(self, request): - data = await self.seller.principals.get_principal(IDENTITY, request) + data = await self.seller.principals.get_principal(IDENTITY, request, context=BETA_2) return TaskResult(status=TaskStatus.COMPLETED, data=data, success=True) async def get_reporting_status(self, request): account_id = request.account.account_id - coverage = await self.seller.engine.subscription_coverage(IDENTITY, account_id) + coverage = await self.seller.engine.subscription_coverage( + IDENTITY, account_id, adcp_version="3.3-beta.2" + ) data = GetReportingStatusResponse.model_construct(subscription_coverage=coverage or None) return TaskResult(status=TaskStatus.COMPLETED, data=data, success=True) @@ -242,3 +247,33 @@ async def test_tracker_keeps_every_registered_version() -> None: "2026-10-03T04:00:00Z", ] assert {p.delivery_config_version for p in periods} == {1} + + +@pytest.mark.asyncio +async def test_regrant_within_retention_keeps_authorized_at() -> None: + # adcp#8585: a re-grant inside retention continues the chain, so + # list_accounts shows the original authorized_at. The tracker records the + # re-grant when it observes the account readable again. + seller = Seller() + client = Client(seller) + tracker = ReportingSubscriptionTracker(client, offerings=[DAILY], status_retention_days=400) + granted = at("2026-09-01T00:00:00Z") + seller.accounts["acct-1"] = AuthorizedAccount("acct-1", timezone=NY, authorized_at=granted) + tracker.observe_account("acct-1", authorized_at=granted, timezone=NY) + await tracker.sync([SUBSCRIPTION]) + await seller.reconcile("2026-10-01T00:00:00Z") + del seller.accounts["acct-1"] + await seller.reconcile("2026-10-04T08:00:00Z") + await tracker.handle_notification(seller.notifications[-1]) + seller.accounts["acct-1"] = AuthorizedAccount("acct-1", timezone=NY, authorized_at=granted) + await seller.reconcile("2026-10-06T15:00:00Z") + await tracker.handle_notification(seller.notifications[-1]) + with pytest.raises(ValueError): + tracker.observe_account("acct-1", authorized_at=granted, timezone=NY) + tracker.observe_account( + "acct-1", authorized_at=granted, timezone=NY, observed_at=at("2026-10-06T15:00:00Z") + ) + (coverage,) = await load_subscription_coverage(client, account={"account_id": "acct-1"}) + now = at("2026-10-10T00:00:00Z") + buyer = tracker.expected_periods("analytics", "acct-1", buys=[BUY], now=now, coverage=coverage) + assert buyer == await seller.engine.owed_periods(IDENTITY, "analytics", "acct-1", now=now) From 1cd9b0e90a882db8795f3bc4c0f94c368e7d9305 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 19:04:03 +0000 Subject: [PATCH 3/4] fix(reporting): address the buyer-tracker review of reporting subscriptions Fixes from the expert review of ReportingSubscriptionTracker: - authorized_at is the chain start (adcp#8585), so any change in it starts a new chain, even when the tracker never saw the loss. The unseen loss is then placed as early as the facts allow: the last instant the account was seen readable, or authorized_at minus retention. The tracker under-expects rather than claim periods the seller never owed. An earlier authorized_at is refused. - expected_periods applies the coverage entry's timezone_cutovers. When the seller lists them, they replace the tracker's own late observations. - observe_account requires an IANA timezone; the SDK never substitutes UTC. - A re-grant within retention is dated from the principal.changed notification's changed_at when the account leaves accounts_unauthorized[], not from poll time. - After a restart, prior versions come from the readback's prior_versions[].configuration (adcp#8588). A missing one is an error unless require_prior_configurations is false. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/consumer.py | 151 +++++++++++++----- tests/test_reporting_subscription_consumer.py | 114 ++++++++++++- 2 files changed, 218 insertions(+), 47 deletions(-) diff --git a/src/adcp/reporting/subscriptions/consumer.py b/src/adcp/reporting/subscriptions/consumer.py index afdd80fc9..fc0dc8ffc 100644 --- a/src/adcp/reporting/subscriptions/consumer.py +++ b/src/adcp/reporting/subscriptions/consumer.py @@ -24,17 +24,18 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass, field -from datetime import datetime +from datetime import datetime, timedelta from typing import Any, Protocol from uuid import uuid4 -from adcp.reporting.subscriptions._clock import utc_instant +from adcp.reporting.subscriptions._clock import require_zone, utc_instant from adcp.reporting.subscriptions.coverage import ( AccountTimezoneChange, ExpectedSubscriptionPeriod, SubscriptionAuthorization, SubscriptionBuy, expected_subscription_periods, + subscription_version_windows, ) from adcp.reporting.subscriptions.resolution import SubscriptionOffering from adcp.types import ( @@ -213,6 +214,10 @@ class _AccountRecord: operator: str | None = None grants: list[SubscriptionAuthorization] = field(default_factory=list) timezone_changes: list[AccountTimezoneChange] = field(default_factory=list) + #: The last instant the account was seen readable. + last_seen: datetime | None = None + #: The chain start ``list_accounts`` last showed as ``authorized_at``. + chain_start: datetime | None = None class ReportingSubscriptionTracker: @@ -295,37 +300,71 @@ async def sync( self._readback = readback return readback - async def refresh(self) -> ReportingSubscriptionReadback: - """Re-read the readback and fold ``accounts_unauthorized[]`` into grant history.""" + async def refresh(self, *, changed_at: datetime | None = None) -> ReportingSubscriptionReadback: + """Re-read the readback and fold ``accounts_unauthorized[]`` into grant history. + + ``changed_at`` is the ``principal.changed`` notification's + ``changed_at``. When an account the tracker saw lose authority leaves + ``accounts_unauthorized[]`` within retention, that is a re-grant the + wire does not otherwise date (adcp#8585 keeps ``authorized_at`` at the + chain start), so it is recorded at ``changed_at``. + """ self._readback = await read_reporting_subscriptions(self._client) + listed: set[str] = set() for state in self._readback.subscriptions: for entry in state.accounts_unauthorized or (): + listed.add(entry.account_id) self._record_loss(entry.account_id, utc_instant(entry.unauthorized_at)) + if changed_at is not None: + at = utc_instant(changed_at, field="changed_at") + for account_id, record in self._accounts.items(): + last = record.grants[-1] if record.grants else None + if last is None or last.unauthorized_at is None or account_id in listed: + continue + expired = self._retention is not None and at - last.unauthorized_at >= timedelta( + days=self._retention + ) + if not expired and at > last.unauthorized_at: + record.grants.append(SubscriptionAuthorization(at)) return self._readback async def handle_notification(self, notification: Mapping[str, Any] | Any) -> bool: """Refresh on a relevant ``principal.changed``; returns whether it refreshed.""" if not is_subscription_change(notification): return False - await self.refresh() + payload = _dump(notification) + changed = payload.get("changed_at") + if isinstance(changed, str): + changed = datetime.fromisoformat(changed.replace("Z", "+00:00")) + await self.refresh(changed_at=changed if isinstance(changed, datetime) else None) return True def observe_account( self, account_id: str, *, - authorized_at: datetime | None, - timezone: str = "UTC", + authorized_at: datetime, + timezone: str, operator: str | None = None, observed_at: datetime | None = None, ) -> None: """Record what ``list_accounts`` shows for an account the caller can read. - After a recorded loss, a later ``authorized_at`` is a re-grant that - started a new chain. An unchanged ``authorized_at`` is a re-grant within - retention (adcp#8585), recorded at ``observed_at``. A changed - ``timezone`` is recorded as a timezone change at ``observed_at``. + ``timezone`` is required: the SDK never substitutes a zone. + ``authorized_at`` is the start of the caller's current authorization + chain (adcp#8585), so any change starts a new chain, even when the + loss before it was never seen; the unseen loss is then placed as early + as the facts allow (the last instant the account was seen readable), + so the tracker under-expects rather than claims periods the seller + never owed. An unchanged ``authorized_at`` after a recorded loss is a + re-grant within retention; :meth:`handle_notification` dates it from + the ``principal.changed`` notification, and ``observed_at`` is only a + fallback. A changed ``timezone`` is a fallback too: the seller's + ``timezone_cutovers[]`` are authoritative in :meth:`expected_periods`. """ + require_zone(timezone, field="timezone") + granted = utc_instant(authorized_at, field="authorized_at") + seen = utc_instant(observed_at, field="observed_at") if observed_at else granted record = self._accounts.get(account_id) if record is None: record = self._accounts[account_id] = _AccountRecord(timezone, operator) @@ -333,39 +372,53 @@ def observe_account( if observed_at is None: raise ValueError("a timezone change needs observed_at") record.timezone_changes.append( - AccountTimezoneChange( - utc_instant(observed_at), timezone, previous_timezone=record.timezone - ) + AccountTimezoneChange(seen, timezone, previous_timezone=record.timezone) ) record.timezone = timezone if operator is not None: record.operator = operator - if authorized_at is None: - return - granted = utc_instant(authorized_at, field="authorized_at") - if not record.grants: + if record.chain_start is None: + record.chain_start = granted record.grants.append(SubscriptionAuthorization(granted)) + record.last_seen = seen return - last = record.grants[-1] - if last.unauthorized_at is None: - return - if granted > last.unauthorized_at: - # A new chain: the seller reset authorized_at to the re-grant. - record.grants.append(SubscriptionAuthorization(granted)) - return - # adcp#8585: a re-grant within status_retention_days continues the - # chain and leaves authorized_at unchanged, so the wire carries no - # re-grant instant. Record the instant this account was observed - # readable again; the seller's gaps[] (when it does not backfill) - # remain authoritative for the periods that are not owed. - if observed_at is None: + if granted < record.chain_start: raise ValueError( - "the account is readable again with an unchanged authorized_at; " - "pass observed_at so the re-grant can be recorded" + f"authorized_at for {account_id!r} moved earlier, which a chain start cannot" ) - record.grants.append( - SubscriptionAuthorization(max(utc_instant(observed_at), last.unauthorized_at)) - ) + last = record.grants[-1] + if granted > record.chain_start: + # A new chain starting at ``granted``. + if last.unauthorized_at is None and last.authorized_at <= record.chain_start: + # Its loss was never seen: end the old grant at the earliest + # instant the facts allow, then start the new chain. + ended = record.last_seen or last.authorized_at + if self._retention is not None: + ended = min(ended, granted - timedelta(days=self._retention)) + ended = max(ended, last.authorized_at + timedelta(microseconds=1)) + record.grants[-1] = SubscriptionAuthorization(last.authorized_at, ended) + record.grants.append(SubscriptionAuthorization(granted)) + elif last.unauthorized_at is None: + # The re-grant is already recorded (dated by a notification); + # the wire's chain start is authoritative for when it began. + record.grants[-1] = SubscriptionAuthorization( + max(granted, record.grants[-2].unauthorized_at or granted) + if len(record.grants) > 1 + else granted + ) + else: + record.grants.append(SubscriptionAuthorization(granted)) + record.chain_start = granted + elif last.unauthorized_at is not None: + # The same chain readable again: a re-grant within retention not + # yet dated by a notification. + if observed_at is None: + raise ValueError( + "the account is readable again with an unchanged authorized_at; " + "pass observed_at, or let handle_notification date the re-grant" + ) + record.grants.append(SubscriptionAuthorization(max(seen, last.unauthorized_at))) + record.last_seen = max(record.last_seen or seen, seen) def _record_loss(self, account_id: str, unauthorized_at: datetime) -> None: record = self._accounts.get(account_id) @@ -384,11 +437,18 @@ def expected_periods( now: datetime, coverage: ReportingSubscriptionCoverage | None = None, include_unresolved: bool = False, + require_prior_configurations: bool = True, ) -> list[ExpectedSubscriptionPeriod]: """What the seller owes this account under the subscription, from buyer records. ``coverage`` is the subscription's ``subscription_coverage[]`` entry - for the account, read while the caller is authorized. + for the account, read while the caller is authorized. Its + ``excluded``, ``gaps`` and ``timezone_cutovers`` all apply; when it + lists timezone cutovers they replace the tracker's own observations. + A prior version whose configuration neither the readback + (adcp#8588) nor this tracker knows is an error unless + ``require_prior_configurations`` is false, which assumes the current + configuration. """ state = self._readback.get(subscription_id) if state is None: @@ -396,22 +456,29 @@ def expected_periods( record = self._accounts.get(account_id) if record is None: raise KeyError(account_id) - return expected_subscription_periods( - subscription=state, + windows = subscription_version_windows( + state, version_configurations=self._sent.get(subscription_id), + require_prior_configurations=require_prior_configurations, + ) + return expected_subscription_periods( + subscription=windows, account_id=account_id, buys=buys, offerings=self._offerings, account_timezone=record.timezone, now=now, authorizations=list(record.grants), - exceptions=() if coverage is None else coverage.excluded, - gaps=() if coverage is None else (coverage.gaps or ()), + coverage=coverage, backfill_on_regrant=self._backfill, status_retention_days=self._retention, destinations=self._destinations, account_operator=record.operator, - timezone_changes=list(record.timezone_changes), + timezone_changes=( + () + if coverage is not None and coverage.timezone_cutovers + else list(record.timezone_changes) + ), include_unresolved=include_unresolved, ) diff --git a/tests/test_reporting_subscription_consumer.py b/tests/test_reporting_subscription_consumer.py index 7218bf485..7e81ec169 100644 --- a/tests/test_reporting_subscription_consumer.py +++ b/tests/test_reporting_subscription_consumer.py @@ -81,7 +81,11 @@ def __init__(self) -> None: async def emit(change: PrincipalChange) -> None: self.notifications.append( - {"notification_type": "principal.changed", "reason": change.reason} + { + "notification_type": "principal.changed", + "reason": change.reason, + "changed_at": change.changed_at.isoformat(), + } ) self.principals = PrincipalService(emit_change=emit) @@ -231,7 +235,9 @@ async def test_tracker_keeps_every_registered_version() -> None: seller = Seller() client = Client(seller) tracker = ReportingSubscriptionTracker(client, offerings=[DAILY]) - seller.accounts["acct-1"] = AuthorizedAccount("acct-1", timezone=NY) + seller.accounts["acct-1"] = AuthorizedAccount( + "acct-1", timezone=NY, authorized_at=at("2026-09-01T00:00:00Z") + ) tracker.observe_account("acct-1", authorized_at=at("2026-09-01T00:00:00Z"), timezone=NY) await tracker.sync([SUBSCRIPTION]) seller.now = at("2026-10-03T12:00:00Z") @@ -267,13 +273,111 @@ async def test_regrant_within_retention_keeps_authorized_at() -> None: await tracker.handle_notification(seller.notifications[-1]) seller.accounts["acct-1"] = AuthorizedAccount("acct-1", timezone=NY, authorized_at=granted) await seller.reconcile("2026-10-06T15:00:00Z") + # The notification's changed_at dates the re-grant; a later poll of + # list_accounts (unchanged authorized_at) adds nothing. await tracker.handle_notification(seller.notifications[-1]) - with pytest.raises(ValueError): - tracker.observe_account("acct-1", authorized_at=granted, timezone=NY) tracker.observe_account( - "acct-1", authorized_at=granted, timezone=NY, observed_at=at("2026-10-06T15:00:00Z") + "acct-1", authorized_at=granted, timezone=NY, observed_at=at("2026-10-09T00:00:00Z") + ) + (coverage,) = await load_subscription_coverage(client, account={"account_id": "acct-1"}) + now = at("2026-10-10T00:00:00Z") + buyer = tracker.expected_periods("analytics", "acct-1", buys=[BUY], now=now, coverage=coverage) + assert buyer == await seller.engine.owed_periods(IDENTITY, "analytics", "acct-1", now=now) + + +@pytest.mark.asyncio +async def test_a_changed_authorized_at_is_a_new_chain_even_if_the_loss_was_missed() -> None: + seller = Seller() + seller.engine._retention = timedelta(days=2) + seller.engine._retention_days = 2 + client = Client(seller) + tracker = ReportingSubscriptionTracker(client, offerings=[DAILY], status_retention_days=2) + granted = at("2026-09-01T00:00:00Z") + seller.accounts["acct-1"] = AuthorizedAccount("acct-1", timezone=NY, authorized_at=granted) + tracker.observe_account("acct-1", authorized_at=granted, timezone=NY) + await tracker.sync([SUBSCRIPTION]) + await seller.reconcile("2026-10-01T00:00:00Z") + tracker.observe_account( + "acct-1", authorized_at=granted, timezone=NY, observed_at=at("2026-10-02T12:00:00Z") + ) + # Lost on 10-03; the tracker misses every notification. + del seller.accounts["acct-1"] + await seller.reconcile("2026-10-03T00:00:00Z") + regrant = at("2026-10-08T00:00:00Z") + seller.accounts["acct-1"] = AuthorizedAccount("acct-1", timezone=NY, authorized_at=regrant) + await seller.reconcile("2026-10-08T00:00:00Z") + tracker.observe_account( + "acct-1", authorized_at=regrant, timezone=NY, observed_at=at("2026-10-08T01:00:00Z") + ) + await tracker.refresh() + now = at("2026-10-12T05:00:00Z") + seller_owes = await seller.engine.owed_periods(IDENTITY, "analytics", "acct-1", now=now) + buyer = tracker.expected_periods("analytics", "acct-1", buys=[BUY], now=now) + owed = {(p.period_start, p.offering_id) for p in seller_owes} + expected = {(p.period_start, p.offering_id) for p in buyer} + # Never a period the seller does not owe: no false obligation_missing. + assert expected <= owed + assert "2026-10-09T04:00:00Z" in {start for start, _ in expected} + + +@pytest.mark.asyncio +async def test_expected_periods_follow_the_sellers_timezone_cutovers() -> None: + seller = Seller() + client = Client(seller) + tracker = ReportingSubscriptionTracker(client, offerings=[DAILY], status_retention_days=400) + granted = at("2026-09-01T00:00:00Z") + seller.accounts["acct-1"] = AuthorizedAccount("acct-1", timezone=NY, authorized_at=granted) + tracker.observe_account("acct-1", authorized_at=granted, timezone=NY) + await tracker.sync([SUBSCRIPTION]) + await seller.reconcile("2026-10-01T00:00:00Z") + seller.accounts["acct-1"] = AuthorizedAccount( + "acct-1", timezone="America/Los_Angeles", authorized_at=granted + ) + await seller.reconcile("2026-10-03T12:00:00Z") + # The tracker only notices the new zone days later. + tracker.observe_account( + "acct-1", + authorized_at=granted, + timezone="America/Los_Angeles", + observed_at=at("2026-10-06T12:00:00Z"), ) (coverage,) = await load_subscription_coverage(client, account={"account_id": "acct-1"}) now = at("2026-10-10T00:00:00Z") buyer = tracker.expected_periods("analytics", "acct-1", buys=[BUY], now=now, coverage=coverage) assert buyer == await seller.engine.owed_periods(IDENTITY, "analytics", "acct-1", now=now) + + +def test_observe_account_requires_a_timezone() -> None: + tracker = ReportingSubscriptionTracker(Client(Seller()), offerings=[DAILY]) + with pytest.raises(TypeError): + tracker.observe_account("acct-1", authorized_at=at("2026-09-01T00:00:00Z")) # type: ignore[call-arg] + with pytest.raises(ValueError): + tracker.observe_account( + "acct-1", authorized_at=at("2026-09-01T00:00:00Z"), timezone="posixrules" + ) + + +@pytest.mark.asyncio +async def test_a_restarted_tracker_reads_prior_configurations_from_the_readback() -> None: + seller = Seller() + client = Client(seller) + first = ReportingSubscriptionTracker(client, offerings=[DAILY]) + granted = at("2026-09-01T00:00:00Z") + seller.accounts["acct-1"] = AuthorizedAccount("acct-1", timezone=NY, authorized_at=granted) + await first.sync([SUBSCRIPTION]) + seller.now = at("2026-10-03T12:00:00Z") + await first.sync( + [{**SUBSCRIPTION, "subscription_version": 2, "scope": {"account_ids": ["acct-2"]}}] + ) + # A new tracker with no local version records. + restarted = ReportingSubscriptionTracker(client, offerings=[DAILY]) + restarted.observe_account("acct-1", authorized_at=granted, timezone=NY) + await restarted.refresh() + periods = restarted.expected_periods( + "analytics", "acct-1", buys=[BUY], now=at("2026-10-08T00:00:00Z") + ) + assert [p.period_start for p in periods] == [ + "2026-10-01T04:00:00Z", + "2026-10-02T04:00:00Z", + "2026-10-03T04:00:00Z", + ] From 5dd4a7a0ac7e9a461c9dd0b1924ad8d1531814c0 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 00:43:42 +0000 Subject: [PATCH 4/4] fix(reporting): refuse a coverage entry for another subscription ReportingSubscriptionTracker.expected_periods applied whatever coverage entry it was given. A buyer with several subscriptions that passed the wrong subscription_coverage[] entry would silently apply that subscription's gaps and timezone cutovers to this one. The tracker now raises when coverage.subscription_id differs from the subscription asked about. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/consumer.py | 7 +++++++ tests/test_reporting_subscription_consumer.py | 18 ++++++++++++++++++ 2 files changed, 25 insertions(+) diff --git a/src/adcp/reporting/subscriptions/consumer.py b/src/adcp/reporting/subscriptions/consumer.py index fc0dc8ffc..a1d46bfb0 100644 --- a/src/adcp/reporting/subscriptions/consumer.py +++ b/src/adcp/reporting/subscriptions/consumer.py @@ -450,6 +450,13 @@ def expected_periods( ``require_prior_configurations`` is false, which assumes the current configuration. """ + if coverage is not None and coverage.subscription_id != subscription_id: + # Another subscription's gaps and cutovers would silently reshape + # this one's owed periods; fail closed on the mix-up. + raise ValueError( + f"coverage is for subscription {coverage.subscription_id!r}, " + f"not {subscription_id!r}" + ) state = self._readback.get(subscription_id) if state is None: raise KeyError(subscription_id) diff --git a/tests/test_reporting_subscription_consumer.py b/tests/test_reporting_subscription_consumer.py index 7e81ec169..c25a2e591 100644 --- a/tests/test_reporting_subscription_consumer.py +++ b/tests/test_reporting_subscription_consumer.py @@ -381,3 +381,21 @@ async def test_a_restarted_tracker_reads_prior_configurations_from_the_readback( "2026-10-02T04:00:00Z", "2026-10-03T04:00:00Z", ] + + +@pytest.mark.asyncio +async def test_coverage_for_another_subscription_is_refused() -> None: + seller = Seller() + client = Client(seller) + tracker = ReportingSubscriptionTracker(client, offerings=[DAILY]) + granted = at("2026-09-01T00:00:00Z") + seller.accounts["acct-1"] = AuthorizedAccount("acct-1", timezone=NY, authorized_at=granted) + tracker.observe_account("acct-1", authorized_at=granted, timezone=NY) + await tracker.sync([SUBSCRIPTION]) + await seller.reconcile("2026-10-01T00:00:00Z") + (coverage,) = await load_subscription_coverage(client, account={"account_id": "acct-1"}) + other = coverage.model_copy(update={"subscription_id": "pacing"}) + with pytest.raises(ValueError, match="pacing"): + tracker.expected_periods( + "analytics", "acct-1", buys=[BUY], now=at("2026-10-05T00:00:00Z"), coverage=other + )