From 05ee4d4bab97092284989d79d4fdefa2eaa49ab9 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 19:33:11 +0000 Subject: [PATCH 1/3] feat(reporting): make the grant journal the one source of authorized_at The reporting-subscription engine now journals every grant it learns of and is the only source of authorization.authorized_at (DX plan slice P6). - The chain rule lives in one place, authorization_chain_start: a re-grant before unauthorized_at + status_retention_days continues the chain, so authorized_at does not move; a later re-grant starts a new chain. Coverage uses the same rule. - On the managed path, served list_accounts and sync_accounts responses carry the journal's authorized_at, the value coverage uses. Each account they show is first observed by the journal: a new grant, a re-grant, or a reduction below get_reporting_status, which reads as a loss. A host value is adopted on first observation; one that later moves backwards is rejected, logged as authorized_at_regressed and counted. Absence from a response never revokes. - reporting.grants.changed(principal, account_id, at=, authorized=, allowed_tasks=) pushes a grant change into the journal and re-evaluates without listing, so a lagging listing cannot undo it. engine.authorization_changed gains allowed_tasks and source. - Grant checks read the journal, which these request-path observations and pushes keep current, so a reduction the caller was shown is enforced on its next reporting task (the staleness #1728 left). - wind_down_grant(current=None) and REPORTING_WIND_DOWN_TASKS (adcp.server) return the reduced grant that ends a relationship without cutting reconciliation, keeping the current authorized_at, as in adcp-client#3543. Journal entries record their source (listed, pushed); listed entries omit it, so existing state documents stay byte-identical. The shared grant-cases.json vectors from adcp-client#3543 run through the Python engine. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reliable-reporting-service.md | 18 ++ src/adcp/decisioning/handler.py | 13 +- src/adcp/reporting/service.py | 23 +- src/adcp/reporting/subscriptions/__init__.py | 5 + src/adcp/reporting/subscriptions/coverage.py | 17 + src/adcp/reporting/subscriptions/engine.py | 193 ++++++++++-- src/adcp/reporting/subscriptions/managed.py | 109 +++++++ src/adcp/server/__init__.py | 3 + src/adcp/server/reporting_grants.py | 57 ++++ .../reporting_subscriptions/grant-cases.json | 171 ++++++++++ ...st_reporting_subscription_grant_journal.py | 294 ++++++++++++++++++ ...test_reporting_subscription_state_store.py | 3 +- 12 files changed, 879 insertions(+), 27 deletions(-) create mode 100644 src/adcp/server/reporting_grants.py create mode 100644 tests/fixtures/reporting_subscriptions/grant-cases.json create mode 100644 tests/test_reporting_subscription_grant_journal.py diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index 0ec8143c9..8deda3b1d 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -379,6 +379,24 @@ platform = reporting.install(MySeller()) # then serve the platform as usual platform that cannot list accounts (`SUBSCRIPTION_ROSTER_UNAVAILABLE`). Both are `ReliableReportingConfigurationError` with `code` and `recovery="correctable"`, and each message ends with the fix. +- **Grants and `authorized_at`.** The SDK journals every grant it learns of and + is the only source of `authorization.authorized_at`. It applies the chain + rule: a re-grant before `unauthorized_at + status_retention_days` continues + the chain, so `authorized_at` does not move; a later re-grant starts a new + one. Served `list_accounts` and `sync_accounts` responses carry the + journal's value, the same one coverage uses. A host value is adopted on + first observation; a later one that moves backwards is rejected and logged + as `authorized_at_regressed`. Those responses also update the journal, as + does `await reporting.grants.changed(principal, account_id, at=..., + authorized=..., allowed_tasks=...)` for sellers whose entitlement service + emits events. Grant checks read the journal, so a reduction shown to the + caller is enforced on its next reporting task. +- **Wind down, don't revoke.** Revocation means no access at all, receipts and + statuses included. To end a relationship while the buyer finishes + reconciling, set `account.authorization = wind_down_grant(account.authorization)` + (from `adcp.server`). It lists only `REPORTING_WIND_DOWN_TASKS` + (`get_reporting_status`, `sync_reporting_receipts`, `sync_reporting_status`) + and keeps `authorized_at`, so coverage continues. - **Principals.** Reporting APIs name a principal with `PrincipalRef(tenant_id, principal_id)`; `grants_task` and `authorizes_delivery` take it. One service serves one tenant. diff --git a/src/adcp/decisioning/handler.py b/src/adcp/decisioning/handler.py index f5e505cd0..1ba5038e0 100644 --- a/src/adcp/decisioning/handler.py +++ b/src/adcp/decisioning/handler.py @@ -3314,6 +3314,9 @@ async def sync_accounts( # type: ignore[override] if inspect.isawaitable(result): result = await result projected = _project_sync_accounts(result) + reporting = getattr(self._platform, "_reliable_reporting_service", None) + if reporting is not None: + projected = await reporting._project_accounts(projected, tool_ctx) await self._reporting_subscription_write(None, tool_ctx) return cast("SyncAccountsResponse", projected) @@ -3354,9 +3357,13 @@ async def list_accounts( # type: ignore[override] from adcp.reporting.outbox.activity import omit_account_activity projected = omit_account_activity(projected) - return cast( - "ListAccountsResponse", strip_credentials_from_wire_result("list_accounts", projected) - ) + served = strip_credentials_from_wire_result("list_accounts", projected) + reporting = getattr(self._platform, "_reliable_reporting_service", None) + if reporting is not None: + # Managed reporting subscriptions serve the grant journal's + # authorized_at, the value coverage uses. + served = await reporting._project_accounts(served, tool_ctx) + return cast("ListAccountsResponse", served) # ----- Optional-method gate ----- diff --git a/src/adcp/reporting/service.py b/src/adcp/reporting/service.py index 20346d649..335b2e7ab 100644 --- a/src/adcp/reporting/service.py +++ b/src/adcp/reporting/service.py @@ -778,6 +778,20 @@ async def touch(self, *, account_id: str | None = None, context: Any | None = No managed = self._require_managed() await managed.touch(account_id=account_id, context=context) + @property + def grants(self) -> Any: + """Push grant changes into the grant journal (``await reporting.grants.changed(...)``). + + See :class:`~adcp.reporting.subscriptions.managed.SubscriptionGrants`. + """ + return self._require_managed().grants + + async def _project_accounts(self, response: Any, context: Any | None) -> Any: + """Serve the grant journal's ``authorized_at`` on an account response.""" + if self._managed is None: + return response + return await self._managed.project_accounts(response, context) + def _require_managed(self) -> ManagedSubscriptions: if self._managed is None: raise ReliableReportingConfigurationError( @@ -1986,7 +2000,12 @@ def triggered(name: str) -> Callable[..., Any] | None: async def write(instance: Any, params: Any, context: Any | None = None) -> Any: result = await _resolve(method(instance, params, context)) - await service._subscription_write(_written_account(params, context), context) + if name in {"sync_accounts", "list_accounts"}: + result = await service._project_accounts(result, context) + if name != "list_accounts": + await service._subscription_write( + _written_account(params, context), context + ) return result write.__name__ = name @@ -1997,7 +2016,7 @@ async def write(instance: Any, params: Any, context: Any | None = None) -> Any: "get_principal": get_principal, "sync_principal": sync_principal, } - for name in ("sync_accounts", "create_media_buy", "update_media_buy"): + for name in ("list_accounts", "sync_accounts", "create_media_buy", "update_media_buy"): wrapped = triggered(name) if wrapped is not None: methods[name] = wrapped diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index e5fa4e9aa..8847af538 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -54,6 +54,7 @@ SubscriptionBuyPackage, SubscriptionException, SubscriptionVersionWindow, + authorization_chain_start, expected_subscription_periods, subscription_covers_account, subscription_exceptions, @@ -113,8 +114,12 @@ project_subscription_state, serves_settled_subscription_fields, ) +from adcp.server.reporting_grants import REPORTING_WIND_DOWN_TASKS, wind_down_grant __all__ = [ + "REPORTING_WIND_DOWN_TASKS", + "authorization_chain_start", + "wind_down_grant", "COVERAGE_REASON_CODES", "COVERAGE_RULE_ORDER", "COVERAGE_VOCABULARY_VERSION", diff --git a/src/adcp/reporting/subscriptions/coverage.py b/src/adcp/reporting/subscriptions/coverage.py index 0d64300df..276d2371f 100644 --- a/src/adcp/reporting/subscriptions/coverage.py +++ b/src/adcp/reporting/subscriptions/coverage.py @@ -575,6 +575,22 @@ class _Owed: media_buy_ids: set[str] = field(default_factory=set) +def authorization_chain_start( + authorizations: Sequence[SubscriptionAuthorization], *, status_retention_days: int +) -> datetime | None: + """The ``authorization.authorized_at`` a seller serves for a grant history. + + The start of the current authority chain: a re-grant before + ``unauthorized_at + status_retention_days`` continues the chain, so the + value never moves on such a re-grant; a later re-grant starts a new chain. + ``None`` while the caller holds no grant. The same rule coverage applies. + """ + if not authorizations or authorizations[-1].unauthorized_at is not None: + return None + chains = _chains(authorizations, timedelta(days=status_retention_days)) + return chains[-1].start + + def _chains( authorizations: Sequence[SubscriptionAuthorization], retention: timedelta | None ) -> list[_Chain]: @@ -1037,6 +1053,7 @@ def _subscription_id(subscription: Any, windows: Sequence[SubscriptionVersionWin "SubscriptionBuyPackage", "SubscriptionException", "SubscriptionVersionWindow", + "authorization_chain_start", "expected_subscription_periods", "subscription_covers_account", "subscription_exceptions", diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index e7d918da3..6af881f41 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -37,7 +37,7 @@ import asyncio import json import logging -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Collection, Mapping, Sequence from dataclasses import dataclass, field, replace from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Any, Protocol, TypeVar @@ -52,6 +52,7 @@ SubscriptionBuy, SubscriptionException, SubscriptionVersionWindow, + authorization_chain_start, expected_subscription_periods, subscription_covers_account, subscription_version_windows, @@ -224,6 +225,11 @@ class _Account: timezone_changes: list[AccountTimezoneChange] = field(default_factory=list) #: The current grant's ``allowed_tasks``; ``None`` when the host does not scope tasks. allowed_tasks: frozenset[str] | None = None + #: How each entry of ``grants`` was learned: ``listed`` (the host's + #: account listing, at reconcile or on a served ``list_accounts``/ + #: ``sync_accounts``), ``pushed`` (``grants.changed``) or + #: ``request_account``. Aligned with ``grants``. + sources: list[str] = field(default_factory=list) def grants_task(self, task: str) -> bool: return self.authorized and (self.allowed_tasks is None or task in self.allowed_tasks) @@ -476,8 +482,11 @@ def encode_principal_state( { "authorized_at": iso_z(grant.authorized_at), "unauthorized_at": _iso(grant.unauthorized_at), + # Absent for listed grants, so earlier documents stay + # byte-identical. + **({"source": source} if source != "listed" else {}), } - for grant in account.grants + for grant, source in zip(account.grants, _grant_sources(account), strict=True) ], "timezone_changes": [ { @@ -551,6 +560,7 @@ def decode_principal_state(document: Mapping[str, Any]) -> _PrincipalState: ) for grant in raw["grants"] ], + sources=[str(grant.get("source", "listed")) for grant in raw["grants"]], timezone_changes=[ AccountTimezoneChange( _required_instant(change["changed_at"]), @@ -604,6 +614,16 @@ def decode_principal_state(document: Mapping[str, Any]) -> _PrincipalState: return principal +def _grant_sources(account: _Account) -> list[str]: + sources = list(account.sources[: len(account.grants)]) + return sources + ["listed"] * (len(account.grants) - len(sources)) + + +def _covers(allowed_tasks: frozenset[str] | None) -> bool: + """Coverage follows the grant: absent task scoping, or ``get_reporting_status`` listed.""" + return allowed_tasks is None or REPORTING_STATUS_TASK in allowed_tasks + + def _dump(value: Any) -> dict[str, Any]: if hasattr(value, "model_dump"): dumped = value.model_dump(mode="json", exclude_none=True) @@ -659,6 +679,7 @@ def __init__( self._store = store or InMemoryReportingSubscriptionStore() self._clock = clock or (lambda: datetime.now(timezone.utc)) self._errors: dict[tuple[str, str], Exception] = {} + self._regressions = 0 if end_account_configs is not None and not callable( getattr(host, "account_delivery_configs", None) ): @@ -677,6 +698,11 @@ def evaluation_errors(self) -> Mapping[tuple[str, str], Exception]: """ return dict(self._errors) + @property + def authorized_at_regressions(self) -> int: + """Host ``authorized_at`` values rejected for moving behind the journal (this process).""" + return self._regressions + @property def support(self) -> ReportingSubscriptionSupport: return self._support @@ -745,6 +771,67 @@ def change(state: _PrincipalState) -> None: await self._mutate(principal.subject, change) + async def authorized_at(self, principal: PrincipalLike, account_id: str) -> datetime | None: + """The ``authorization.authorized_at`` to serve: the current chain's start. + + From the grant journal, the same history coverage reads, so the value a + buyer reads on ``list_accounts`` is the value the engine used. ``None`` + while the principal holds no grant on the account. + """ + account = (await self._read(principal.subject)).accounts.get(account_id) + return None if account is None else self._chain_start(account) + + async def observe_grant( + self, + principal: PrincipalLike, + account_id: str, + *, + allowed_tasks: Collection[str] | None, + authorized_at: datetime | None = None, + timezone: str | None = None, + operator: str | None = None, + at: datetime | None = None, + ) -> datetime | None: + """Record a grant the host served on the request path; return ``authorized_at``. + + For an account a served ``list_accounts`` or ``sync_accounts`` shows + with its authorization: the journal records a new grant, a re-grant or + a reduction below ``get_reporting_status`` (a loss), adopts the host's + ``authorized_at`` on first observation, and rejects one that moves + backwards. Absence from a response never revokes: responses are paged + and filtered. Returns the chain start to serve, or ``None``. + """ + instant = self._now(at) + tasks = None if allowed_tasks is None else frozenset(str(item) for item in allowed_tasks) + covered = _covers(tasks) + reported = None if authorized_at is None else utc_instant(authorized_at) + if timezone is not None: + require_zone(timezone, field="timezone") + + def change(state: _PrincipalState) -> datetime | None: + account = state.accounts.get(account_id) + if account is None: + if not covered: + return None + account = state.accounts[account_id] = _Account( + timezone=timezone, operator=operator + ) + self._observe(account, covered=covered, reported=reported, at=instant, source="listed") + account.allowed_tasks = tasks if covered else None + account.set_timezone(timezone, instant) + if operator is not None: + account.operator = operator + return self._chain_start(account) + + return await self._mutate(principal.subject, change, now=instant) + + async def refresh(self, principal: PrincipalLike, *, now: datetime | None = None) -> None: + """Re-evaluate exceptions from the journal without listing the host's accounts. + + For pushed grant changes: a listing that lags the push must not undo it. + """ + await self._evaluate(principal, self._now(now)) + async def account_ids(self, principal: PrincipalLike) -> frozenset[str]: """Every account the engine holds state for under the principal.""" return frozenset((await self._read(principal.subject)).accounts) @@ -855,33 +942,99 @@ async def authorization_changed( at: datetime, timezone: str | None = None, operator: str | None = None, + allowed_tasks: Collection[str] | None = None, + source: str = "pushed", ) -> None: - """Record a grant or a loss of authority at ``at``.""" + """Record a grant, a change to it, or a loss of authority at ``at``. + + ``allowed_tasks`` is the grant's task list (``None`` restricts no + task). A grant that does not list ``get_reporting_status`` reads as a + loss of authority at ``at``, exactly like a revocation; a wind-down + grant that lists it keeps coverage. A re-grant inside + ``status_retention_days`` continues the chain, so the projected + ``authorized_at`` does not move. + """ instant = utc_instant(at, field="at") if timezone is not None: require_zone(timezone, field="timezone") + tasks = None if allowed_tasks is None else frozenset(str(item) for item in allowed_tasks) + covered = authorized and _covers(tasks) def change(principal: _PrincipalState) -> None: account = principal.accounts.get(account_id) if account is None: - if not authorized: + if not covered: return account = principal.accounts[account_id] = _Account( timezone=timezone, operator=operator ) - self._apply_grant(account, authorized=authorized, at=instant) + self._observe(account, covered=covered, reported=None, at=instant, source=source) + account.allowed_tasks = tasks if covered else None account.set_timezone(timezone, instant) if operator is not None: account.operator = operator await self._mutate(identity.subject, change, now=instant) + def _observe( + self, + account: _Account, + *, + covered: bool, + reported: datetime | None, + at: datetime, + source: str, + ) -> bool: + """Apply one observation of a grant to the journal; ``True`` if it regressed. + + A first grant adopts the host's ``authorized_at`` (never later than + the observation). A re-grant is dated by a reported value later than + the loss, else by the observation. While the grant holds, a reported + value earlier than the journal's chain start is a regression: it is + rejected, logged and counted, never adopted. + """ + if not covered: + if account.authorized: + self._apply_grant(account, authorized=False, at=at) + return False + if not account.authorized: + lost = account.grants[-1].unauthorized_at if account.grants else None + # adcp#8585: authorized_at is the chain start. A later one is a new + # grant dated then; an unchanged one is a re-grant within + # retention, which the wire does not date, so it is stamped when + # observed. + if reported is not None and (lost is None or reported > lost): + granted = min(reported, at) + else: + granted = at + self._apply_grant(account, authorized=True, at=granted, source=source) + return False + start = self._chain_start(account) + if reported is not None and start is not None and reported < start: + self._regressions += 1 + logger.warning( + "authorized_at_regressed: the host reported authorized_at %s, before the " + "chain start %s the grant journal holds; the journal keeps its value", + iso_z(reported), + iso_z(start), + extra={"event": "reporting.subscription.authorized_at_regressed"}, + ) + return True + return False + + def _chain_start(self, account: _Account) -> datetime | None: + return authorization_chain_start(account.grants, status_retention_days=self._retention_days) + @staticmethod - def _apply_grant(account: _Account, *, authorized: bool, at: datetime) -> None: + def _apply_grant( + account: _Account, *, authorized: bool, at: datetime, source: str = "listed" + ) -> 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.sources = _grant_sources(account) account.grants.append(SubscriptionAuthorization(at)) + account.sources.append(source) elif not authorized and account.authorized: last = account.grants[-1] account.grants[-1] = SubscriptionAuthorization( @@ -916,7 +1069,7 @@ async def reconcile(self, identity: PrincipalLike, *, now: datetime | None = Non listed = { item.account_id: item for item in await self._list_accounts(identity) - if item.allowed_tasks is None or REPORTING_STATUS_TASK in item.allowed_tasks + if _covers(item.allowed_tasks) } def change(principal: _PrincipalState) -> None: @@ -926,18 +1079,13 @@ def change(principal: _PrincipalState) -> 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) + self._observe( + account, + covered=True, + reported=item.authorized_at, + at=instant, + source="listed", + ) account.set_timezone(item.timezone, instant) account.operator = item.operator account.allowed_tasks = item.allowed_tasks @@ -1745,8 +1893,11 @@ async def frozen_owed_sets( async def grants_task(self, identity: PrincipalLike, account_id: str, task: str) -> bool: """Whether the principal's current grant on the account allows ``task``. - From the engine's last reconcile of the host's grants. There is no - window after a loss of authority: a revoked account allows nothing. + From the grant journal: the last reconcile of the host's grants, kept + current by request-path observations (a served ``list_accounts`` or + ``sync_accounts``) and pushed changes (``authorization_changed``). + There is no window after a loss of authority: a revoked account allows + nothing. """ account = (await self._read(identity.subject)).accounts.get(account_id) return account is not None and account.grants_task(task) diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py index 36e55a16c..0b5adc646 100644 --- a/src/adcp/reporting/subscriptions/managed.py +++ b/src/adcp/reporting/subscriptions/managed.py @@ -418,6 +418,56 @@ class _Platform: sync_accounts_seen: set[str] = field(default_factory=set) +class SubscriptionGrants: + """Push grant changes into the grant journal: ``reporting.grants``. + + For sellers whose entitlement service emits events: a pushed change is + dated exactly, where a listing can only observe it. The journal then + re-evaluates the principal's exceptions without listing the host's + accounts, so a listing that lags the push cannot undo it. Listings + observed later must agree with what was pushed. + """ + + def __init__(self, managed: ManagedSubscriptions) -> None: + self._managed = managed + + async def changed( + self, + principal: Any, + account_id: str, + *, + at: datetime, + authorized: bool, + allowed_tasks: Sequence[str] | None = None, + timezone: str | None = None, + operator: str | None = None, + ) -> None: + """Record that ``principal``'s grant on ``account_id`` changed at ``at``. + + ``authorized=False`` is a revocation: no access at all from ``at``. + ``allowed_tasks`` is the new grant's task list (``None`` restricts no + task); a grant without ``get_reporting_status`` ends coverage exactly + like a revocation, and :func:`~adcp.server.wind_down_grant`'s tasks + keep it. A re-grant inside ``status_retention_days`` continues the + chain, so the ``authorized_at`` served on ``list_accounts`` does not + move. + """ + managed = self._managed + ref = PrincipalRef.coerce(principal) + await managed.engine.authorization_changed( + ref, + account_id, + authorized=authorized, + at=at, + timezone=timezone, + operator=operator, + allowed_tasks=allowed_tasks, + source="pushed", + ) + if await managed._subscribed(ref): + await managed.engine.refresh(ref) + + class ManagedSubscriptions: """The SDK-owned composition behind ``ReliableReportingService(subscriptions=...)``. @@ -446,6 +496,7 @@ def __init__( #: Served-write triggers waiting for the drain: (account_id, caller). self._queued: dict[tuple[str | None, str | None], PrincipalRef | None] = {} self._drain: asyncio.Task[None] | None = None + self.grants = SubscriptionGrants(self) # -- composition ----------------------------------------------------------- @@ -936,6 +987,63 @@ def _buy(self, raw: Mapping[str, Any]) -> SubscriptionBuy | None: ), ) + # -- authorized_at projection ---------------------------------------------- + + async def project_accounts(self, response: Any, context: Any) -> Any: + """Serve the journal's ``authorized_at`` on a ``list_accounts``/``sync_accounts`` response. + + Each account shown with an authorization is first observed by the + journal (a new grant, a re-grant, a reduction, or a regressed + ``authorized_at``, which is rejected), so grant checks read what the + caller was just shown. The served ``authorized_at`` is the value + coverage uses. Only callers holding subscriptions are touched, and a + failure here never fails the response. + """ + if not self.composed: + return response + metadata = getattr(context, "metadata", None) + if isinstance(metadata, Mapping) and metadata.get("adcp.reporting.service_context"): + return response + payload = _wire(response) + if not isinstance(payload, Mapping) or not isinstance(payload.get("accounts"), list): + return response + try: + principal = await self.resolve(None, context) + if not await self._subscribed(principal): + return response + accounts: list[Any] = [] + for raw in payload["accounts"]: + accounts.append(await self._project_account(principal, raw)) + except Exception: # noqa: BLE001 - the projection never fails the read + logger.exception("reporting subscription authorized_at projection failed") + return response + return {**payload, "accounts": accounts} + + async def _project_account(self, principal: PrincipalRef, raw: Any) -> Any: + from adcp.reporting.subscriptions._clock import iso_z + + if not isinstance(raw, Mapping): + return raw + authorization = raw.get("authorization") + account_id = raw.get("account_id") + if ( + not account_id + or not isinstance(authorization, Mapping) + or authorization.get("allowed_tasks") is None + ): + return raw + start = await self.engine.observe_grant( + principal, + str(account_id), + allowed_tasks=[str(item) for item in authorization["allowed_tasks"]], + authorized_at=_instant(authorization.get("authorized_at")), + timezone=raw.get("timezone") or None, + operator=raw.get("operator"), + ) + if start is None: + return raw + return {**raw, "authorization": {**authorization, "authorized_at": iso_z(start)}} + # -- capabilities ---------------------------------------------------------- @staticmethod @@ -961,6 +1069,7 @@ def advertise_principal(payload: dict[str, Any]) -> dict[str, Any]: __all__ = [ + "SubscriptionGrants", "AccountConfigsOverride", "AccountsOverride", "BuyOfferingsOverride", diff --git a/src/adcp/server/__init__.py b/src/adcp/server/__init__.py index c260732e8..ed0508e3f 100644 --- a/src/adcp/server/__init__.py +++ b/src/adcp/server/__init__.py @@ -160,6 +160,7 @@ async def get_products(params, context=None): principal_record_to_document, ) from adcp.server.proposal import ProposalBuilder, ProposalNotSupported +from adcp.server.reporting_grants import REPORTING_WIND_DOWN_TASKS, wind_down_grant from adcp.server.responses import ( activate_signal_response, build_creative_response, @@ -254,6 +255,8 @@ async def get_products(params, context=None): "PrincipalChangeEmitter", "PrincipalIdentity", "PrincipalRef", + "REPORTING_WIND_DOWN_TASKS", + "wind_down_grant", "PrincipalRecord", "PrincipalRecordStore", "PrincipalService", diff --git a/src/adcp/server/reporting_grants.py b/src/adcp/server/reporting_grants.py new file mode 100644 index 000000000..7229bc1e2 --- /dev/null +++ b/src/adcp/server/reporting_grants.py @@ -0,0 +1,57 @@ +"""Grants for reporting subscriptions: wind down instead of revoking (AdCP 3.3, experimental). + +Revocation means no access (adcp#8607): once a principal's grant on an +account is gone, every task is refused, ``get_reporting_status``, receipts and +statuses included. A seller that wants a buyer to finish reconciling keeps a +reduced grant instead, listing only the reconciliation tasks. Coverage +continues while the grant lists ``get_reporting_status``, so periods already +owed are still delivered and can still be acknowledged. + +The TypeScript SDK spells these ``windDownGrant`` and +``REPORTING_WIND_DOWN_TASKS``. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +#: The tasks a wind-down grant keeps: reading status, and the two writes a +#: buyer needs to finish reconciling. +REPORTING_WIND_DOWN_TASKS: tuple[str, ...] = ( + "get_reporting_status", + "sync_reporting_receipts", + "sync_reporting_status", +) + + +def wind_down_grant(current: Any = None) -> Any: + """The reduced grant that ends a relationship without cutting reconciliation. + + Returns an :class:`~adcp.types.AccountAuthorization` listing only + :data:`REPORTING_WIND_DOWN_TASKS`, with ``read_only`` false because + receipts and statuses are writes. Pass the account's current + authorization (a model or a mapping) to keep its ``authorized_at``: a + reduced grant continues the authorization chain, and sellers advertising + reporting subscriptions must keep serving it. The old ``field_scopes`` and + ``scope_name`` describe the old task set, so they are dropped:: + + account.authorization = wind_down_grant(account.authorization) + """ + from adcp.types import AccountAuthorization + + payload: dict[str, Any] = { + "allowed_tasks": list(REPORTING_WIND_DOWN_TASKS), + "read_only": False, + } + authorized_at = ( + current.get("authorized_at") + if isinstance(current, Mapping) + else getattr(current, "authorized_at", None) + ) + if authorized_at is not None: + payload["authorized_at"] = authorized_at + return AccountAuthorization.model_validate(payload) + + +__all__ = ["REPORTING_WIND_DOWN_TASKS", "wind_down_grant"] diff --git a/tests/fixtures/reporting_subscriptions/grant-cases.json b/tests/fixtures/reporting_subscriptions/grant-cases.json new file mode 100644 index 000000000..1923bc9c4 --- /dev/null +++ b/tests/fixtures/reporting_subscriptions/grant-cases.json @@ -0,0 +1,171 @@ +{ + "description": "Reduced grants and reporting-subscription coverage (experimental media_buy.reporting_subscriptions, AdCP 3.3). Revocation means no access (adcp#8607): a seller that wants a buyer to finish reconciling keeps a reduced grant instead of revoking. Coverage follows the grant: an account is covered while its authorization, when advertised, lists get_reporting_status in allowed_tasks. A grant without get_reporting_status reads exactly as a loss of authority at the instant it took effect.", + "rule": "covered = authorization is absent, or authorization.allowed_tasks lists get_reporting_status. read_only does not affect coverage.", + "spec_refs": [ + "static/schemas/source/core/reporting-subscription.json#/x-adcp-validation/coverage", + "docs/media-buy/task-reference/get_reporting_status.mdx (Winding down instead of revoking)", + "static/schemas/source/media-buy/sync-reporting-receipts-request.json#/x-adcp-validation/after_revocation" + ], + "wind_down_grant": { + "allowed_tasks": ["get_reporting_status", "sync_reporting_receipts", "sync_reporting_status"], + "read_only": false + }, + "grant_cases": [ + { "name": "no authorization object: restricts no task", "authorization": null, "covered": true }, + { + "name": "wind-down grant", + "authorization": { + "allowed_tasks": ["get_reporting_status", "sync_reporting_receipts", "sync_reporting_status"], + "read_only": false + }, + "covered": true + }, + { + "name": "get_reporting_status alone", + "authorization": { "allowed_tasks": ["get_reporting_status"], "read_only": true }, + "covered": true + }, + { + "name": "full buying grant listing get_reporting_status", + "authorization": { + "allowed_tasks": ["get_media_buys", "create_media_buy", "get_reporting_status", "sync_reporting_receipts"] + }, + "covered": true + }, + { + "name": "reduced grant without get_reporting_status: receipts and statuses only", + "authorization": { "allowed_tasks": ["sync_reporting_receipts", "sync_reporting_status"], "read_only": false }, + "covered": false + }, + { + "name": "reduced grant without get_reporting_status: statuses only", + "authorization": { "allowed_tasks": ["sync_reporting_status"], "read_only": false }, + "covered": false + }, + { + "name": "reduced grant without get_reporting_status: buying tasks only", + "authorization": { "allowed_tasks": ["get_media_buys", "create_media_buy"] }, + "covered": false + }, + { + "name": "empty allowed_tasks", + "authorization": { "allowed_tasks": [], "read_only": true }, + "covered": false + }, + { + "name": "task names match exactly", + "authorization": { "allowed_tasks": ["get_reporting_status_v2", "GET_REPORTING_STATUS"] }, + "covered": false + } + ], + "transition_conventions": { + "offering": "One delivery offering: period_duration P1D, alignment utc, delivery_sla PT0S, feed_purpose analytics, finality snapshot, delivery_only.", + "subscription": "analytics-daily version 1, scope all_authorized_accounts, accepted at 2026-09-01T12:00:00Z.", + "buy": "mb-1 on account-a, confirmed 2026-09-01T06:00:00Z, flight 2026-09-02T00:00:00Z to 2026-09-11T00:00:00Z, one package mapped to the offering.", + "grant": "authorized_at 2026-08-01T00:00:00Z. Each step replaces the account's authorization at its instant; the seller reconciles at that instant and the revocation instant is the observation time. allowed_tasks null means an authorization that restricts no task; revoked means the account leaves the list.", + "expected": "Read at evaluate_at: authorized is the engine's current authority, covered is whether get_reporting_status shows the subscription on the account, unauthorized_at lists accounts_unauthorized for account-a, owed lists every owed period start in order." + }, + "transition_cases": [ + { + "name": "wind-down keeps coverage and opens no accounts_unauthorized entry", + "steps": [ + { "at": "2026-09-01T18:00:00Z", "allowed_tasks": null }, + { + "at": "2026-09-05T12:00:00Z", + "allowed_tasks": ["get_reporting_status", "sync_reporting_receipts", "sync_reporting_status"] + } + ], + "evaluate_at": "2026-09-13T00:00:00Z", + "expected": { + "authorized": true, + "covered": true, + "unauthorized_at": [], + "owed": [ + "2026-09-02T00:00:00Z", + "2026-09-03T00:00:00Z", + "2026-09-04T00:00:00Z", + "2026-09-05T00:00:00Z", + "2026-09-06T00:00:00Z", + "2026-09-07T00:00:00Z", + "2026-09-08T00:00:00Z", + "2026-09-09T00:00:00Z", + "2026-09-10T00:00:00Z" + ] + } + }, + { + "name": "a reduced grant without get_reporting_status ends coverage like a revocation", + "steps": [ + { "at": "2026-09-01T18:00:00Z", "allowed_tasks": null }, + { "at": "2026-09-05T12:00:00Z", "allowed_tasks": ["sync_reporting_receipts", "sync_reporting_status"] } + ], + "evaluate_at": "2026-09-13T00:00:00Z", + "expected": { + "authorized": false, + "covered": false, + "unauthorized_at": ["2026-09-05T12:00:00Z"], + "owed": ["2026-09-02T00:00:00Z", "2026-09-03T00:00:00Z", "2026-09-04T00:00:00Z"] + } + }, + { + "name": "a reduced grant without get_reporting_status owes the same periods as a full revocation", + "steps": [ + { "at": "2026-09-01T18:00:00Z", "allowed_tasks": null }, + { "at": "2026-09-05T12:00:00Z", "allowed_tasks": "revoked" } + ], + "evaluate_at": "2026-09-13T00:00:00Z", + "expected": { + "authorized": false, + "covered": false, + "unauthorized_at": ["2026-09-05T12:00:00Z"], + "owed": ["2026-09-02T00:00:00Z", "2026-09-03T00:00:00Z", "2026-09-04T00:00:00Z"] + } + }, + { + "name": "dropping get_reporting_status from a wind-down grant ends coverage then", + "steps": [ + { "at": "2026-09-01T18:00:00Z", "allowed_tasks": null }, + { + "at": "2026-09-04T12:00:00Z", + "allowed_tasks": ["get_reporting_status", "sync_reporting_receipts", "sync_reporting_status"] + }, + { "at": "2026-09-07T12:00:00Z", "allowed_tasks": ["sync_reporting_receipts"] } + ], + "evaluate_at": "2026-09-13T00:00:00Z", + "expected": { + "authorized": false, + "covered": false, + "unauthorized_at": ["2026-09-07T12:00:00Z"], + "owed": [ + "2026-09-02T00:00:00Z", + "2026-09-03T00:00:00Z", + "2026-09-04T00:00:00Z", + "2026-09-05T00:00:00Z", + "2026-09-06T00:00:00Z" + ] + } + }, + { + "name": "restoring get_reporting_status inside retention is a re-grant with a gap", + "steps": [ + { "at": "2026-09-01T18:00:00Z", "allowed_tasks": null }, + { "at": "2026-09-04T12:00:00Z", "allowed_tasks": ["sync_reporting_receipts", "sync_reporting_status"] }, + { "at": "2026-09-06T12:00:00Z", "allowed_tasks": null, "granted_at": "2026-09-06T12:00:00Z" } + ], + "evaluate_at": "2026-09-13T00:00:00Z", + "expected": { + "authorized": true, + "covered": true, + "unauthorized_at": [], + "owed": [ + "2026-09-02T00:00:00Z", + "2026-09-03T00:00:00Z", + "2026-09-07T00:00:00Z", + "2026-09-08T00:00:00Z", + "2026-09-09T00:00:00Z", + "2026-09-10T00:00:00Z" + ] + } + } + ] +} diff --git a/tests/test_reporting_subscription_grant_journal.py b/tests/test_reporting_subscription_grant_journal.py new file mode 100644 index 000000000..1ad4ccc8d --- /dev/null +++ b/tests/test_reporting_subscription_grant_journal.py @@ -0,0 +1,294 @@ +"""The grant journal: the one source of ``authorization.authorized_at`` (AdCP 3.3). + +The engine journals every grant it learns of (listed, pushed), applies the +chain rule, serves the chain start on ``list_accounts``/``sync_accounts``, and +feeds the same history to coverage. Grant checks read the journal, which +request-path observations and pushed changes keep current. Revocation means no +access (adcp#8607); ``wind_down_grant`` keeps reconciliation. + +``grant-cases.json`` is shared with the TypeScript SDK (adcp-client#3543). +""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +import pytest + +from adcp.exceptions import ADCPTaskError +from adcp.reporting import PrincipalRef +from adcp.reporting.subscriptions import ( + AuthorizedAccount, + ReportingSubscriptionEngine, + SubscriptionAuthorization, + SubscriptionBuy, + SubscriptionBuyPackage, + SubscriptionOffering, + authorization_chain_start, +) +from adcp.server import ( + REPORTING_WIND_DOWN_TASKS, + PrincipalIdentity, + PrincipalService, + wind_down_grant, +) +from tests.test_reporting_subscription_engine import Clock, Host +from tests.test_reporting_subscription_managed import ( + ACCOUNT, + AGENCY, + START, + _ctx, + _World, +) + +VECTORS = json.loads( + (Path(__file__).parent / "fixtures/reporting_subscriptions/grant-cases.json").read_text() +) +IDENTITY = PrincipalIdentity("https://agency.example/agent", "buyer_agent") +DAILY = SubscriptionOffering( + offering_id="daily", + report_definition_id="def-daily", + reporting_profile="delivery", + feed_purpose="analytics", + period_duration="P1D", + alignment="utc", + delivery_sla="PT0S", + supported_finality=frozenset({"snapshot"}), + reconciliation_mode="delivery_only", + dimensions=frozenset({"package"}), +) + + +def at(text: str) -> datetime: + return datetime.fromisoformat(text.replace("Z", "+00:00")) + + +# -- shared vectors ------------------------------------------------------------ + + +def test_wind_down_grant_matches_the_shared_vector() -> None: + assert list(REPORTING_WIND_DOWN_TASKS) == VECTORS["wind_down_grant"]["allowed_tasks"] + assert ( + wind_down_grant().model_dump(mode="json", exclude_none=True) == VECTORS["wind_down_grant"] + ) + kept = wind_down_grant( + {"allowed_tasks": ["create_media_buy"], "authorized_at": "2026-08-01T00:00:00Z"} + ).model_dump(mode="json", exclude_none=True) + assert kept["authorized_at"] == "2026-08-01T00:00:00Z" + assert kept["allowed_tasks"] == list(REPORTING_WIND_DOWN_TASKS) + + +@pytest.mark.parametrize("case", VECTORS["grant_cases"], ids=lambda case: case["name"]) +async def test_coverage_follows_the_grant(case: dict[str, Any]) -> None: + engine, host, principals = await _engine() + authorization = case["authorization"] + tasks = None if authorization is None else authorization["allowed_tasks"] + host.accounts["account-a"] = AuthorizedAccount( + "account-a", + timezone="UTC", + authorized_at=at("2026-08-01T00:00:00Z"), + allowed_tasks=None if tasks is None else frozenset(tasks), + ) + await engine.reconcile(IDENTITY) + assert ( + await engine.grants_task(IDENTITY, "account-a", "get_reporting_status") is case["covered"] + ) + + +async def _engine() -> tuple[ReportingSubscriptionEngine, Host, PrincipalService]: + clock = Clock("2026-09-01T12:00:00Z") + host = Host() + principals = PrincipalService() + engine = ReportingSubscriptionEngine( + principals=principals, host=host, offerings=[DAILY], clock=clock + ) + principals.set_reporting_subscriptions(engine.principal_options()) + response = await principals.sync_principal( + IDENTITY, + { + "idempotency_key": "grant-journal-key-000001", + "configuration": { + "reporting_subscriptions": [ + { + "subscription_id": "analytics-daily", + "subscription_version": 1, + "active": True, + "scope": {"all_authorized_accounts": True}, + "selector": { + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "snapshot", + "dimensions": ["package"], + }, + "reconciliation_mode": "delivery_only", + } + ] + }, + }, + ) + assert response.result.kind == "applied" + host.buys["account-a"] = [ + SubscriptionBuy( + media_buy_id="mb-1", + confirmed_at=at("2026-09-01T06:00:00Z"), + start_time=at("2026-09-02T00:00:00Z"), + end_time=at("2026-09-11T00:00:00Z"), + packages=(SubscriptionBuyPackage("pkg-1", ("daily",)),), + ) + ] + return engine, host, principals + + +@pytest.mark.parametrize("case", VECTORS["transition_cases"], ids=lambda case: case["name"]) +async def test_grant_transitions_through_the_engine(case: dict[str, Any]) -> None: + engine, host, principals = await _engine() + authorized_at = at("2026-08-01T00:00:00Z") + for step in case["steps"]: + tasks = step["allowed_tasks"] + if tasks == "revoked": + host.accounts.pop("account-a", None) + else: + host.accounts["account-a"] = AuthorizedAccount( + "account-a", + timezone="UTC", + authorized_at=at(step.get("granted_at", "2026-08-01T00:00:00Z")) + if "granted_at" in step + else authorized_at, + allowed_tasks=None if tasks is None else frozenset(tasks), + ) + await engine.reconcile(IDENTITY, now=at(step["at"])) + expected = case["expected"] + now = at(case["evaluate_at"]) + await engine.reconcile(IDENTITY, now=now) + assert ( + await engine.grants_task(IDENTITY, "account-a", "get_reporting_status") + is expected["authorized"] + ) + owed = await engine.owed_periods(IDENTITY, "analytics-daily", "account-a", now=now) + assert [at(period.period_start) for period in owed] == [at(item) for item in expected["owed"]] + [state] = await principals.reporting_subscriptions(IDENTITY) + unauthorized = [ + item.unauthorized_at.strftime("%Y-%m-%dT%H:%M:%SZ") + for item in state.accounts_unauthorized or [] + if item.account_id == "account-a" + ] + assert unauthorized == expected["unauthorized_at"] + + +# -- the chain rule ------------------------------------------------------------ + + +def test_the_chain_start_holds_inside_retention_and_restarts_after_it() -> None: + first = SubscriptionAuthorization(at("2026-01-01T00:00:00Z"), at("2026-03-01T00:00:00Z")) + inside = [first, SubscriptionAuthorization(at("2026-03-20T00:00:00Z"))] + assert authorization_chain_start(inside, status_retention_days=30) == at("2026-01-01T00:00:00Z") + after = [first, SubscriptionAuthorization(at("2026-04-15T00:00:00Z"))] + assert authorization_chain_start(after, status_retention_days=30) == at("2026-04-15T00:00:00Z") + assert authorization_chain_start([first], status_retention_days=30) is None + assert authorization_chain_start([], status_retention_days=30) is None + + +# -- the managed path ---------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_list_accounts_serves_the_journals_authorized_at() -> None: + world = _World() + del world.seller.accounts[ACCOUNT]["authorization"]["authorized_at"] + await world.subscribe() + listed = await world.platform.list_accounts({}, _ctx()) + served = listed["accounts"][0]["authorization"]["authorized_at"] + # The host omitted it: the journal dated the grant at its first observation. + assert served == "2026-11-01T00:00:00Z" + engine = world.service._managed.engine + assert await engine.authorized_at(PrincipalRef(None, AGENCY), ACCOUNT) == START + + +@pytest.mark.asyncio +async def test_a_host_value_that_moves_backwards_is_rejected() -> None: + world = _World() + await world.subscribe() + engine = world.service._managed.engine + world.seller.accounts[ACCOUNT]["authorization"]["authorized_at"] = "2026-09-01T00:00:00Z" + listed = await world.platform.list_accounts({}, _ctx()) + assert listed["accounts"][0]["authorization"]["authorized_at"] == "2026-10-01T00:00:00Z" + assert engine.authorized_at_regressions == 1 + + +@pytest.mark.asyncio +async def test_a_regrant_inside_retention_keeps_authorized_at() -> None: + world = _World() + await world.subscribe() + principal = PrincipalRef(None, AGENCY) + grants = world.service.grants + await grants.changed(principal, ACCOUNT, at=START + timedelta(days=2), authorized=False) + await grants.changed(principal, ACCOUNT, at=START + timedelta(days=3), authorized=True) + engine = world.service._managed.engine + assert await engine.authorized_at(principal, ACCOUNT) == datetime( + 2026, 10, 1, tzinfo=timezone.utc + ) + world.clock.set(START + timedelta(days=3)) + listed = await world.platform.list_accounts({}, _ctx()) + assert listed["accounts"][0]["authorization"]["authorized_at"] == "2026-10-01T00:00:00Z" + + +@pytest.mark.asyncio +async def test_a_pushed_revocation_takes_effect_at_once() -> None: + world = _World() + await world.subscribe() + principal = PrincipalRef(None, AGENCY) + revoked_at = START + timedelta(minutes=10) + world.clock.set(revoked_at) + await world.service.grants.changed(principal, ACCOUNT, at=revoked_at, authorized=False) + engine = world.service._managed.engine + assert not await engine.grants_task(principal, ACCOUNT, "sync_reporting_receipts") + assert await engine.authorized_at(principal, ACCOUNT) is None + # The readback records the loss without waiting for a listing. + [state] = await world.service.principals.reporting_subscriptions(principal) + assert [item.account_id for item in state.accounts_unauthorized] == [ACCOUNT] + + +@pytest.mark.asyncio +async def test_a_wind_down_keeps_reconciliation_open() -> None: + world = _World() + await world.subscribe() + principal = PrincipalRef(None, AGENCY) + await world.service.grants.changed( + principal, + ACCOUNT, + at=START + timedelta(minutes=10), + authorized=True, + allowed_tasks=REPORTING_WIND_DOWN_TASKS, + ) + engine = world.service._managed.engine + for task in REPORTING_WIND_DOWN_TASKS: + assert await engine.grants_task(principal, ACCOUNT, task) + assert not await engine.grants_task(principal, ACCOUNT, "get_media_buys") + assert await engine.authorized_at(principal, ACCOUNT) == datetime( + 2026, 10, 1, tzinfo=timezone.utc + ) + + +@pytest.mark.asyncio +async def test_a_served_reduction_is_read_by_the_next_grant_check() -> None: + """#1728: grant checks no longer wait for the next reconcile.""" + world = _World() + await world.subscribe() + world.clock.set(START + timedelta(hours=3, minutes=30)) + await world.service.run_worker() + world.seller.accounts[ACCOUNT]["authorization"]["allowed_tasks"] = ["get_media_buys"] + await world.platform.list_accounts({}, _ctx()) + with pytest.raises(ADCPTaskError) as refused: + await world.platform.get_reporting_status({"view": "periods"}, _ctx()) + assert refused.value.errors[0].code == "ACCOUNT_NOT_FOUND" + + +@pytest.mark.asyncio +async def test_callers_without_subscriptions_are_served_unchanged() -> None: + world = _World() + listed = await world.platform.list_accounts({}, _ctx("https://stranger.example/agent")) + assert listed["accounts"][0]["authorization"]["authorized_at"] == "2026-10-01T00:00:00Z" + assert await world.service._managed.engine.list_principals() == [] diff --git a/tests/test_reporting_subscription_state_store.py b/tests/test_reporting_subscription_state_store.py index 3f97e6c01..e9d63f695 100644 --- a/tests/test_reporting_subscription_state_store.py +++ b/tests/test_reporting_subscription_state_store.py @@ -162,7 +162,8 @@ async def test_a_conflicting_write_is_reloaded_and_reapplied() -> None: stored = await store.load(IDENTITY.subject) assert stored is not None and stored.revision == 1 assert stored.document["accounts"]["acct-1"]["grants"] == [ - {"authorized_at": "2026-10-01T00:00:00Z", "unauthorized_at": None} + # A pushed grant records its source; listed grants omit it. + {"authorized_at": "2026-10-01T00:00:00Z", "unauthorized_at": None, "source": "pushed"} ] From 5e23835c3b29465f4faee65b97cf3c1d7304549d Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 20:06:32 +0000 Subject: [PATCH 2/3] fix(reporting): observe a served account page in one journal write A subscribed caller's list_accounts or sync_accounts response used to cost one engine-state load and compare-and-swap per account. engine.observe_grants records every grant a page shows in one write, and observe_grant delegates to it. GrantObservation names one account's served grant. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/engine.py | 82 ++++++++++++++----- src/adcp/reporting/subscriptions/managed.py | 60 ++++++++------ ...st_reporting_subscription_grant_journal.py | 24 ++++++ 3 files changed, 121 insertions(+), 45 deletions(-) diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index 6af881f41..63e461eec 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -191,6 +191,17 @@ async def account_delivery_configs( ) -> Sequence[AccountDeliveryConfig]: ... +@dataclass(frozen=True) +class GrantObservation: + """One account's grant as a served ``list_accounts``/``sync_accounts`` showed it.""" + + account_id: str + allowed_tasks: Collection[str] | None + authorized_at: datetime | None = None + timezone: str | None = None + operator: str | None = None + + class AccountConfigsEnder(Protocol): """Ends the caller's own account-level configurations for ``replace_account_configs``. @@ -801,28 +812,60 @@ async def observe_grant( backwards. Absence from a response never revokes: responses are paged and filtered. Returns the chain start to serve, or ``None``. """ - instant = self._now(at) - tasks = None if allowed_tasks is None else frozenset(str(item) for item in allowed_tasks) - covered = _covers(tasks) - reported = None if authorized_at is None else utc_instant(authorized_at) - if timezone is not None: - require_zone(timezone, field="timezone") + observed = await self.observe_grants( + principal, + [GrantObservation(account_id, allowed_tasks, authorized_at, timezone, operator)], + at=at, + ) + return observed.get(account_id) - def change(state: _PrincipalState) -> datetime | None: - account = state.accounts.get(account_id) - if account is None: - if not covered: - return None - account = state.accounts[account_id] = _Account( - timezone=timezone, operator=operator + async def observe_grants( + self, + principal: PrincipalLike, + observations: Sequence[GrantObservation], + *, + at: datetime | None = None, + ) -> dict[str, datetime | None]: + """:meth:`observe_grant` for a whole response page, in one state write. + + Returns ``account_id`` -> the chain start to serve (``None`` without a + grant). One load and at most one compare-and-swap per call. + """ + instant = self._now(at) + prepared = [] + for item in observations: + tasks = ( + None + if item.allowed_tasks is None + else frozenset(str(task) for task in item.allowed_tasks) + ) + if item.timezone is not None: + require_zone(item.timezone, field="timezone") + reported = None if item.authorized_at is None else utc_instant(item.authorized_at) + prepared.append((item, tasks, _covers(tasks), reported)) + + def change(state: _PrincipalState) -> dict[str, datetime | None]: + served: dict[str, datetime | None] = {} + for item, tasks, covered, reported in prepared: + account = state.accounts.get(item.account_id) + if account is None: + if not covered: + served[item.account_id] = None + continue + account = _Account(timezone=item.timezone, operator=item.operator) + state.accounts[item.account_id] = account + self._observe( + account, covered=covered, reported=reported, at=instant, source="listed" ) - self._observe(account, covered=covered, reported=reported, at=instant, source="listed") - account.allowed_tasks = tasks if covered else None - account.set_timezone(timezone, instant) - if operator is not None: - account.operator = operator - return self._chain_start(account) + account.allowed_tasks = tasks if covered else None + account.set_timezone(item.timezone, instant) + if item.operator is not None: + account.operator = item.operator + served[item.account_id] = self._chain_start(account) + return served + if not prepared: + return {} return await self._mutate(principal.subject, change, now=instant) async def refresh(self, principal: PrincipalLike, *, now: datetime | None = None) -> None: @@ -2030,6 +2073,7 @@ def _destination_for( __all__ = [ + "GrantObservation", "AccountDeliveryConfig", "AuthorizedAccount", "AuthorizedAccountPage", diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py index 0b5adc646..79eea3036 100644 --- a/src/adcp/reporting/subscriptions/managed.py +++ b/src/adcp/reporting/subscriptions/managed.py @@ -1011,38 +1011,46 @@ async def project_accounts(self, response: Any, context: Any) -> Any: principal = await self.resolve(None, context) if not await self._subscribed(principal): return response - accounts: list[Any] = [] - for raw in payload["accounts"]: - accounts.append(await self._project_account(principal, raw)) + accounts = await self._project_page(principal, list(payload["accounts"])) except Exception: # noqa: BLE001 - the projection never fails the read logger.exception("reporting subscription authorized_at projection failed") return response return {**payload, "accounts": accounts} - async def _project_account(self, principal: PrincipalRef, raw: Any) -> Any: + async def _project_page(self, principal: PrincipalRef, page: list[Any]) -> list[Any]: + """Observe every shown grant in one journal write, then serve its ``authorized_at``.""" from adcp.reporting.subscriptions._clock import iso_z - - if not isinstance(raw, Mapping): - return raw - authorization = raw.get("authorization") - account_id = raw.get("account_id") - if ( - not account_id - or not isinstance(authorization, Mapping) - or authorization.get("allowed_tasks") is None - ): - return raw - start = await self.engine.observe_grant( - principal, - str(account_id), - allowed_tasks=[str(item) for item in authorization["allowed_tasks"]], - authorized_at=_instant(authorization.get("authorized_at")), - timezone=raw.get("timezone") or None, - operator=raw.get("operator"), - ) - if start is None: - return raw - return {**raw, "authorization": {**authorization, "authorized_at": iso_z(start)}} + from adcp.reporting.subscriptions.engine import GrantObservation + + observations: list[GrantObservation] = [] + for raw in page: + authorization = raw.get("authorization") if isinstance(raw, Mapping) else None + if ( + isinstance(raw, Mapping) + and raw.get("account_id") + and isinstance(authorization, Mapping) + and authorization.get("allowed_tasks") is not None + ): + observations.append( + GrantObservation( + str(raw["account_id"]), + [str(item) for item in authorization["allowed_tasks"]], + _instant(authorization.get("authorized_at")), + raw.get("timezone") or None, + raw.get("operator"), + ) + ) + served = await self.engine.observe_grants(principal, observations) + projected: list[Any] = [] + for raw in page: + start = served.get(str(raw.get("account_id"))) if isinstance(raw, Mapping) else None + if start is None: + projected.append(raw) + continue + authorization = dict(raw["authorization"]) + authorization["authorized_at"] = iso_z(start) + projected.append({**raw, "authorization": authorization}) + return projected # -- capabilities ---------------------------------------------------------- diff --git a/tests/test_reporting_subscription_grant_journal.py b/tests/test_reporting_subscription_grant_journal.py index 1ad4ccc8d..0a562e7d5 100644 --- a/tests/test_reporting_subscription_grant_journal.py +++ b/tests/test_reporting_subscription_grant_journal.py @@ -292,3 +292,27 @@ async def test_callers_without_subscriptions_are_served_unchanged() -> None: listed = await world.platform.list_accounts({}, _ctx("https://stranger.example/agent")) assert listed["accounts"][0]["authorization"]["authorized_at"] == "2026-10-01T00:00:00Z" assert await world.service._managed.engine.list_principals() == [] + + +@pytest.mark.asyncio +async def test_a_served_page_is_observed_in_one_journal_write() -> None: + world = _World() + for index in range(3): + world.seller.accounts[f"account-{index}"] = { + **world.seller.accounts[ACCOUNT], + "account_id": f"account-{index}", + } + await world.subscribe() + store = world.service._managed.engine.store + loads = 0 + original = store.load + + async def counting(subject: str) -> Any: + nonlocal loads + loads += 1 + return await original(subject) + + store.load = counting # type: ignore[method-assign] + listed = await world.platform.list_accounts({}, _ctx()) + assert len(listed["accounts"]) == 4 + assert loads == 1 From 14be0440fbc849e42268310657febe65ee9429f2 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 20:16:56 +0000 Subject: [PATCH 3/3] fix(reporting): observe a served authorization that lists no tasks An authorization without allowed_tasks restricts no task. The reconcile already read it as a full grant; the request-path projection skipped it and served the host's raw authorized_at. It is now observed and projected like any other grant. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/managed.py | 6 ++++-- tests/test_reporting_subscription_grant_journal.py | 12 ++++++++++++ 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py index 79eea3036..217b96836 100644 --- a/src/adcp/reporting/subscriptions/managed.py +++ b/src/adcp/reporting/subscriptions/managed.py @@ -1029,12 +1029,14 @@ async def _project_page(self, principal: PrincipalRef, page: list[Any]) -> list[ isinstance(raw, Mapping) and raw.get("account_id") and isinstance(authorization, Mapping) - and authorization.get("allowed_tasks") is not None ): + tasks = authorization.get("allowed_tasks") observations.append( GrantObservation( str(raw["account_id"]), - [str(item) for item in authorization["allowed_tasks"]], + # An authorization without allowed_tasks restricts no + # task, as the reconcile reads it. + None if tasks is None else [str(item) for item in tasks], _instant(authorization.get("authorized_at")), raw.get("timezone") or None, raw.get("operator"), diff --git a/tests/test_reporting_subscription_grant_journal.py b/tests/test_reporting_subscription_grant_journal.py index 0a562e7d5..c7f30d833 100644 --- a/tests/test_reporting_subscription_grant_journal.py +++ b/tests/test_reporting_subscription_grant_journal.py @@ -316,3 +316,15 @@ async def counting(subject: str) -> Any: listed = await world.platform.list_accounts({}, _ctx()) assert len(listed["accounts"]) == 4 assert loads == 1 + + +@pytest.mark.asyncio +async def test_an_authorization_without_allowed_tasks_is_observed_as_a_full_grant() -> None: + world = _World() + await world.subscribe() + authorization = world.seller.accounts[ACCOUNT]["authorization"] + del authorization["allowed_tasks"] + authorization["authorized_at"] = "2026-09-01T00:00:00Z" # regressed: rejected + listed = await world.platform.list_accounts({}, _ctx()) + assert listed["accounts"][0]["authorization"]["authorized_at"] == "2026-10-01T00:00:00Z" + assert world.service._managed.engine.authorized_at_regressions == 1