From adaba82a4b3449a2b6b62960eb3c9fe6664f43a2 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 19:23:43 +0000 Subject: [PATCH 1/5] feat(reporting): serve reporting subscriptions from one SubscriptionOptions block ReliableReportingService(subscriptions=SubscriptionOptions(...)) is the managed seller path for AdCP 3.3 reporting subscriptions (DX plan slice P5). The block's presence enables the feature and the SDK derives the rest: - the service owns a PrincipalService, links it to the engine, and install() serves get_principal and sync_principal and advertises the principal section, on ADCPHandler, decisioning and production installs; - the roster comes from the platform's list_accounts (a decisioning platform's accounts.list) and the buys from get_media_buys, both read under the principal's identity: a service context carrying its last authenticated binding, recorded at sync_principal with no credential material; - offerings come from the registered capability_offerings, each of which must publish standard_dimensions; - sync_principal, sync_accounts, create_media_buy and update_media_buy reconcile the principals concerned, and each worker turn reconciles every principal a period boundary passed for before planning; service.touch(account_id=...) covers changes the SDK did not serve. Every derivation has an override (accounts, buys, buy_offerings, configure, account_delivery_configs, is_period_delivered, resolve_principal). A listing that fails under the service context raises SUBSCRIPTION_ROSTER_UNAVAILABLE and never reads as an empty roster; startup refuses an offering without standard_dimensions (SUBSCRIPTIONS_REQUIRE_STANDARD_DIMENSIONS) and a platform with no roster. ReliableReportingConfigurationError gains code and recovery for the four seller configuration codes of error-codes.json. replace_account_configs is applied once per subscription_version, or when it changes to true, through end_account_configs and account_delivery_configs; without them it is still refused with UNSUPPORTED_FEATURE. PrincipalRef(tenant_id, principal_id) names a principal across the reporting APIs: grants_task, authorizes_delivery (and the materializer's delivery authority hook) take it. The decisioning Account gains timezone and operator, projected onto the wire. The engine store gains list_principals(), and the engine state document records the binding, the last reconcile and the applied replacements, each only once set. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reliable-reporting-service.md | 62 +- src/adcp/_principal_ref.py | 56 ++ src/adcp/decisioning/account_projection.py | 4 + src/adcp/decisioning/handler.py | 49 +- src/adcp/decisioning/types.py | 8 + src/adcp/reporting/__init__.py | 15 + src/adcp/reporting/_errors.py | 28 + src/adcp/reporting/materializer/service.py | 9 +- src/adcp/reporting/service.py | 254 ++++- src/adcp/reporting/service_subscriptions.py | 24 +- src/adcp/reporting/subscriptions/engine.py | 265 +++++- src/adcp/reporting/subscriptions/managed.py | 890 ++++++++++++++++++ src/adcp/reporting/subscriptions/registry.py | 12 +- src/adcp/server/__init__.py | 2 + src/adcp/server/principal.py | 16 +- tests/fixtures/public_api_snapshot.json | 3 + tests/test_decisioning_handler_shims.py | 3 + ...porting_subscription_delivery_authority.py | 20 +- tests/test_reporting_subscription_managed.py | 529 +++++++++++ tests/test_reporting_subscription_parity.py | 20 + 20 files changed, 2207 insertions(+), 62 deletions(-) create mode 100644 src/adcp/_principal_ref.py create mode 100644 src/adcp/reporting/subscriptions/managed.py create mode 100644 tests/test_reporting_subscription_managed.py diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index 12fbb4b31..05e6017f7 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -324,6 +324,64 @@ contracts, and [the wiring example](../examples/reporting_service_production.py) principal through `sync_principal`, rather than once per account through `sync_accounts`. `adcp.reporting.subscriptions` holds both halves. +**Seller, managed path (recommended).** Add a `subscriptions` block to the +service you already run. Its presence enables the feature; the SDK derives the +rest from what the platform already serves. + +```python +from adcp.reporting import ReliableReportingService, SubscriptionOptions + +reporting = ReliableReportingService( + store=store, + account_context=resolve_account_context, # also receives subscription generations + subscriptions=SubscriptionOptions(max_accounts=500), +) +reporting.sources.register("gam", gam_source, capability_offerings=[daily_offering]) +platform = reporting.install(MySeller()) # then serve the platform as usual +``` + +| The SDK derives | From | Override on `SubscriptionOptions` | +|---|---|---| +| The principal runtime, `get_principal`/`sync_principal` and the principal section of capabilities | the service's own `PrincipalService` | `principals`, `principal_store` | +| The accounts a principal may read | the platform's `list_accounts` (a decisioning platform's `accounts.list`), read under the principal's identity | `accounts(principal, *, cursor)` | +| Each account's buys | the platform's `get_media_buys`, under the same identity | `buys(principal, account_id)` | +| Each package's offerings | every registered offering | `buy_offerings(media_buy)` | +| Offerings | the registered `capability_offerings`, each publishing `standard_dimensions` | none | +| Report content per generation | `account_context` | `configure(generation)` | +| The authenticated principal | the caller the reporting tasks resolve, with `ToolContext.tenant_id` | `resolve_principal(request, context)` | + +- **Identity outside a request.** Reads made for a principal outside a request + carry its last authenticated binding, recorded at `sync_principal`: the + tenant and a copy of its `AuthInfo` with no credential material (key ids, + client ids and verified signers only). A seller whose accounts are only + reachable through a live upstream token supplies `accounts`. A listing that + fails raises `SUBSCRIPTION_ROSTER_UNAVAILABLE` and is reported through + `planning_error_handler`; it never reads as an empty roster, so nothing is + revoked. +- **Accounts must carry `timezone`.** Coverage never substitutes a zone. + `Account(timezone=..., operator=...)` serves both on a decisioning platform. +- **Scheduling.** `sync_principal`, `sync_accounts`, `create_media_buy` and + `update_media_buy` served through the installed platform reconcile the + principals concerned (`reconcile_on_write`, on by default). Each worker turn + first reconciles every principal a period boundary has passed for since its + last reconcile, then plans. Call `await reporting.touch(account_id=...)` after + a change the SDK did not serve, such as a seller UI edit. +- **`replace_account_configs`.** Supply `account_delivery_configs` and + `end_account_configs(principal, account_id, ids, *, at)`. The flag is then + applied once per `subscription_version`, or when it changes to true. + Without them it is refused with `UNSUPPORTED_FEATURE`. +- **Configuration errors.** Startup refuses an offering without + `standard_dimensions` (`SUBSCRIPTIONS_REQUIRE_STANDARD_DIMENSIONS`) and a + platform that cannot list accounts (`SUBSCRIPTION_ROSTER_UNAVAILABLE`). Both + are `ReliableReportingConfigurationError` with `code` and + `recovery="correctable"`, and each message ends with the fix. +- **Principals.** Reporting APIs name a principal with + `PrincipalRef(tenant_id, principal_id)`; `grants_task` and + `authorizes_delivery` take it. One service serves one tenant. + +The rest of this section describes the building blocks the managed path +composes, for sellers who wire them by hand. + **Buyer.** `expected_subscription_periods(...)` is the coverage rule as a pure function over plain data. Pass it the `get_principal` subscription readback, the buys you can see (`SubscriptionBuy`, with earlier flights when a buy was @@ -346,11 +404,11 @@ coverage=...)` then returns what the seller owes. A withheld period carries no `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 +**Seller, advanced composition.** Opt in on the principal runtime and link it to the engine. Without `reporting_subscriptions=`, `PrincipalService` rejects the section with `UNSUPPORTED_FEATURE`, as the spec requires for an unsupported section. `replace_account_configs: true` is also rejected with `UNSUPPORTED_FEATURE` -until the engine can deactivate account-level configurations. +unless the engine is built with `end_account_configs`. ```python from adcp.reporting.subscriptions import ( diff --git a/src/adcp/_principal_ref.py b/src/adcp/_principal_ref.py new file mode 100644 index 000000000..adc58c78e --- /dev/null +++ b/src/adcp/_principal_ref.py @@ -0,0 +1,56 @@ +"""The principal a reporting subscription belongs to (AdCP 3.3, experimental). + +Kept dependency-free so the reporting runtime, the materializer and the server +can all name it without importing each other. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class PrincipalRef: + """The identity a reporting subscription belongs to: ``(tenant_id, principal_id)``. + + ``principal_id`` is the seller-resolved, stable identity of the + authenticated caller (the same value the reporting caller resolver yields + as ``consumer_id``), never anything read from request content. + ``tenant_id`` names the tenant on a multi-tenant server and is ``None`` on + a single-tenant one. One reporting engine serves one tenant: state is keyed + by ``principal_id``, and the tenant is recorded with it. + + The TypeScript SDK spells it ``PrincipalRef { tenantId, principalId }``. + """ + + tenant_id: str | None + principal_id: str + + def __post_init__(self) -> None: + if not isinstance(self.principal_id, str) or not self.principal_id: + raise ValueError("PrincipalRef.principal_id must be a non-empty string") + if self.tenant_id is not None and ( + not isinstance(self.tenant_id, str) or not self.tenant_id + ): + raise ValueError("PrincipalRef.tenant_id must be a non-empty string or None") + + @property + def subject(self) -> str: + """The key reporting state and ledger ownership use (``principal_id``).""" + return self.principal_id + + @classmethod + def coerce(cls, value: Any, *, tenant_id: str | None = None) -> PrincipalRef: + """Accept a ``PrincipalRef``, a ``PrincipalIdentity`` or a principal id string.""" + if isinstance(value, PrincipalRef): + return value + if isinstance(value, str): + return cls(tenant_id, value) + subject = getattr(value, "subject", None) + if isinstance(subject, str): + return cls(getattr(value, "tenant_id", tenant_id), subject) + raise TypeError("expected a PrincipalRef, a PrincipalIdentity or a principal id") + + +__all__ = ["PrincipalRef"] diff --git a/src/adcp/decisioning/account_projection.py b/src/adcp/decisioning/account_projection.py index 418c873e4..3e52fa8b6 100644 --- a/src/adcp/decisioning/account_projection.py +++ b/src/adcp/decisioning/account_projection.py @@ -249,6 +249,10 @@ def to_wire_account(account: DecisioningAccount[Any]) -> dict[str, Any]: authorization = _project_account_authorization(account.authorization) if authorization is not None: wire["authorization"] = authorization + if account.timezone is not None: + wire["timezone"] = account.timezone + if account.operator is not None: + wire["operator"] = account.operator return wire diff --git a/src/adcp/decisioning/handler.py b/src/adcp/decisioning/handler.py index c863e9153..0b1af0f01 100644 --- a/src/adcp/decisioning/handler.py +++ b/src/adcp/decisioning/handler.py @@ -1320,6 +1320,10 @@ class PlatformHandler(ADCPHandler[ToolContext]): "get_reporting_status", "sync_reporting_status", "sync_reporting_receipts", + # Served only when an installed ReliableReportingService serves + # reporting subscriptions on the managed path. + "get_principal", + "sync_principal", "get_task_status", "list_tasks", } @@ -1525,6 +1529,11 @@ def __init__( supported_versions, adcp_version=self._adcp_version ) self._platform = platform + installed_reporting = getattr(platform, "_reliable_reporting_service", None) + if installed_reporting is not None: + # Managed reporting subscriptions read accounts and buys through + # this handler, under each principal's identity. + installed_reporting._bind_handler(self) self._account_activity = account_activity self._reporting_activity = reporting_activity self._reporting_status = reporting_status @@ -2912,6 +2921,7 @@ async def _persist_overlay_hook(create_result: Any) -> None: **webhook_kwargs, ) self._maybe_auto_emit_sync_completion("create_media_buy", params, result) + await self._reporting_subscription_write(account.id, tool_ctx) return cast("CreateMediaBuyResponse", result) async def update_media_buy( # type: ignore[override] @@ -2971,6 +2981,7 @@ async def _merge_overlay_hook(_update_result: Any) -> None: **await self._handoff_webhook_kwargs(params, tool_ctx), ) self._maybe_auto_emit_sync_completion("update_media_buy", params, result) + await self._reporting_subscription_write(account.id, tool_ctx) return cast("UpdateMediaBuySuccessResponse", result) async def sync_creatives( # type: ignore[override] @@ -3060,6 +3071,40 @@ async def _invoke_reporting_method( registry=self._registry, ) + async def _reporting_subscription_write( + self, account_id: str | None, tool_ctx: ToolContext + ) -> None: + """Reconcile managed reporting subscriptions after a served write.""" + reporting = getattr(self._platform, "_reliable_reporting_service", None) + if reporting is not None: + await reporting._subscription_write(account_id, tool_ctx) + + async def _invoke_principal_method( + self, name: str, params: Any, context: ToolContext | None + ) -> Any: + reporting = getattr(self._platform, "_reliable_reporting_service", None) + if reporting is None or name not in reporting.reporting_tools: + return self._not_supported(name) + from adcp.reporting._errors import installed_reporting_call + + tool_ctx = context or ToolContext() + await self._prime_auth_context(tool_ctx) + return await installed_reporting_call( + name, lambda: getattr(reporting, name)(params, tool_ctx), decisioning=True + ) + + async def get_principal( + self, params: Any, context: ToolContext | None = None + ) -> Any: + """Served by an installed ``ReliableReportingService(subscriptions=...)``.""" + return await self._invoke_principal_method("get_principal", params, context) + + async def sync_principal( + self, params: Any, context: ToolContext | None = None + ) -> Any: + """Served by an installed ``ReliableReportingService(subscriptions=...)``.""" + return await self._invoke_principal_method("sync_principal", params, context) + async def get_reporting_status( # type: ignore[override] self, params: GetReportingStatusRequest, context: ToolContext | None = None ) -> Any: @@ -3266,7 +3311,9 @@ async def sync_accounts( # type: ignore[override] result = _call_with_optional_ctx(upsert, refs, ctx=resolve_ctx) if inspect.isawaitable(result): result = await result - return cast("SyncAccountsResponse", _project_sync_accounts(result)) + projected = _project_sync_accounts(result) + await self._reporting_subscription_write(None, tool_ctx) + return cast("SyncAccountsResponse", projected) async def list_accounts( # type: ignore[override] self, diff --git a/src/adcp/decisioning/types.py b/src/adcp/decisioning/types.py index 3eee00efd..b2db5651e 100644 --- a/src/adcp/decisioning/types.py +++ b/src/adcp/decisioning/types.py @@ -478,6 +478,8 @@ class Account(Generic[TMeta]): publisher-identity grants. This is response metadata about the authenticated caller's access, not proof of downstream serving authorization for any individual request. + :param timezone: The account's IANA timezone, emitted as ``timezone``. + :param operator: The account's operator domain, emitted as ``operator``. :param mode: SDK-internal account mode — ``'live'`` (default, production), ``'sandbox'`` (adopter's test infra), or ``'mock'`` (Phase 2 — SDK routes to mock-server backend). @@ -509,6 +511,12 @@ class Account(Generic[TMeta]): rate_card: str | None = None reporting_bucket: ReportingBucket | None = None authorization: AccountAuthorization | dict[str, Any] | None = None + #: The account's IANA timezone (wire ``timezone``). Sellers advertising + #: reporting subscriptions populate it: account-timezone offerings never + #: fall back to UTC. + timezone: str | None = None + #: The operator domain the account belongs to (wire ``operator``). + operator: str | None = None # SDK-internal account mode for sandbox-authority gating. Default # ``'live'`` preserves all existing-adopter behavior — pre-mode diff --git a/src/adcp/reporting/__init__.py b/src/adcp/reporting/__init__.py index ae2c83f1f..de4411c8e 100644 --- a/src/adcp/reporting/__init__.py +++ b/src/adcp/reporting/__init__.py @@ -30,6 +30,7 @@ from typing import TYPE_CHECKING, Any +from adcp._principal_ref import PrincipalRef from adcp.reporting._reconcile import ( ConsumerLoopView, ConsumerStatusCheckpoint, @@ -117,6 +118,12 @@ from adcp.reporting.service_production import ( ReportingServiceOffering as ReportingServiceOffering, ) + from adcp.reporting.service_subscriptions import ( + ReportingSubscriptionServiceOptions as ReportingSubscriptionServiceOptions, + ) + from adcp.reporting.service_subscriptions import ( + SubscriptionOptions as SubscriptionOptions, + ) from adcp.reporting.testing import ( DeterministicReportingClock as DeterministicReportingClock, ) @@ -182,6 +189,11 @@ "ReportingAdapter": ("service", "ReportingAdapter"), "ReportingProductionOptions": ("service_production", "ReportingProductionOptions"), "ReportingServiceOffering": ("service_production", "ReportingServiceOffering"), + "ReportingSubscriptionServiceOptions": ( + "service_subscriptions", + "ReportingSubscriptionServiceOptions", + ), + "SubscriptionOptions": ("service_subscriptions", "SubscriptionOptions"), "DeterministicReportingClock": ("testing", "DeterministicReportingClock"), "FaultInjectingReportingSealStore": ("testing", "FaultInjectingReportingSealStore"), "FaultInjectingReportingStagingStore": ("testing", "FaultInjectingReportingStagingStore"), @@ -237,6 +249,9 @@ def __dir__() -> list[str]: "ReportingAdapter", "ReportingProductionOptions", "ReportingServiceOffering", + "PrincipalRef", + "ReportingSubscriptionServiceOptions", + "SubscriptionOptions", "DeterministicReportingClock", "FaultInjectingReportingSealStore", "FaultInjectingReportingStagingStore", diff --git a/src/adcp/reporting/_errors.py b/src/adcp/reporting/_errors.py index dd5c6e017..848beaf70 100644 --- a/src/adcp/reporting/_errors.py +++ b/src/adcp/reporting/_errors.py @@ -59,21 +59,49 @@ ) +#: Seller configuration codes from the spec's ``error-codes.json`` (adcp#8616). +#: Each is correctable by the seller, and its message names the key or command. +SELLER_CONFIGURATION_CODES = frozenset( + { + "REPORTING_STORAGE_NOT_DURABLE", + "REPORTING_SCHEMA_BEHIND", + "SUBSCRIPTIONS_REQUIRE_STANDARD_DIMENSIONS", + "SUBSCRIPTION_ROSTER_UNAVAILABLE", + } +) + + class ReliableReportingConfigurationError(ValueError): """A composition failure, missing caller, or invalid buyer configuration. Composition failures default to terminal INTERNAL_ERROR. Only explicit request-time failures may ask the buyer to correct authentication/input. + + ``code`` is set for the seller configuration codes of the cross-SDK + vocabulary (``REPORTING_STORAGE_NOT_DURABLE``, ``REPORTING_SCHEMA_BEHIND``, + ``SUBSCRIPTIONS_REQUIRE_STANDARD_DIMENSIONS``, + ``SUBSCRIPTION_ROSTER_UNAVAILABLE``); those carry ``recovery="correctable"`` + and a message whose last sentence is the fix. ``dispatched`` is always + ``None``: a configuration check sends nothing. """ + dispatched = None + def __init__( self, message: str, *, kind: Literal["misconfiguration", "auth_required", "invalid_request"] = "misconfiguration", + code: str | None = None, ) -> None: super().__init__(message) + if code is not None and code not in SELLER_CONFIGURATION_CODES: + raise ValueError(f"unknown seller configuration code {code!r}") self.kind = kind + self.code = code + self.recovery: Literal["correctable", "terminal"] | None = ( + "correctable" if code is not None else None + ) def reporting_error(code: str, message: str) -> Error: diff --git a/src/adcp/reporting/materializer/service.py b/src/adcp/reporting/materializer/service.py index 174f263b8..44c9172a9 100644 --- a/src/adcp/reporting/materializer/service.py +++ b/src/adcp/reporting/materializer/service.py @@ -8,6 +8,7 @@ from datetime import datetime, timedelta, timezone from typing import Any +from adcp._principal_ref import PrincipalRef from adcp.reporting.materializer._errors import materializer_errors from adcp.reporting.materializer.contracts import ( ReportingDestinationWriter, @@ -66,9 +67,9 @@ async def run(self) -> None: return -#: (owning principal subject, account_id) -> whether its current grant still -#: carries the subscription; for example ReportingSubscriptionEngine.authorizes_delivery. -SubscriptionDeliveryAuthority = Callable[[str, str], Awaitable[bool]] +#: (owning principal, account_id) -> whether its current grant still carries the +#: subscription; for example ReportingSubscriptionEngine.authorizes_delivery. +SubscriptionDeliveryAuthority = Callable[[PrincipalRef, str], Awaitable[bool]] @dataclass(frozen=True) @@ -121,7 +122,7 @@ async def _require_subscription_authority( # destination call: a hung hook becomes DEADLINE_EXCEEDED under the # same identity (held), never a lease renewed forever. allowed = await context.run( - lambda: authorize(key.consumer_id, key.account_id), effect=effect + lambda: authorize(PrincipalRef(None, key.consumer_id), key.account_id), effect=effect ) if allowed is not True: raise ReportingWriterError( diff --git a/src/adcp/reporting/service.py b/src/adcp/reporting/service.py index 73532f4be..856a048df 100644 --- a/src/adcp/reporting/service.py +++ b/src/adcp/reporting/service.py @@ -75,6 +75,7 @@ ReportingSubscriptionServiceOptions, SubscriptionPlanningFailure, ) + from adcp.reporting.subscriptions.managed import ManagedSubscriptions, SubscriptionOptions __all__ = [ "AdapterRegistration", @@ -170,6 +171,26 @@ async def _resolve(value: Any) -> Any: return await value if inspect.isawaitable(value) else value +#: Principal tasks the managed subscription path serves. +_MANAGED_TASKS = frozenset({"get_principal", "sync_principal"}) + + +def _written_account(params: Any, context: Any | None) -> str | None: + """The account a served write names, when it names exactly one.""" + account_id = getattr(context, "account_id", None) + if isinstance(account_id, str) and account_id: + return account_id + account = getattr(params, "account", None) + if account is None and isinstance(params, Mapping): + account = params.get("account") + value = getattr(account, "account_id", None) + if value is None and isinstance(account, Mapping): + value = account.get("account_id") + if value is None: + value = getattr(getattr(account, "root", None), "account_id", None) + return value if isinstance(value, str) and value else None + + @dataclass(frozen=True) class ReportingAccountContext: """Trusted source facts frozen for one configuration generation.""" @@ -494,13 +515,19 @@ def __init__( seals: ReportingSealStore | None = None, initialize_source_storage: Callable[[], Awaitable[None]] | None = None, reporting_subscriptions: ReportingSubscriptionServiceOptions | None = None, + subscriptions: SubscriptionOptions | None = None, ) -> None: """``staging``/``seals`` are the default source stores adapters register with; ``initialize_source_storage`` runs once at startup, after ``store.create_schema()``, so a durable storage distribution can install its own schema. - ``reporting_subscriptions`` serves AdCP 3.3 principal-level reporting - subscriptions (see :mod:`adcp.reporting.service_subscriptions`); without - it the service is unchanged.""" + + ``subscriptions`` serves AdCP 3.3 principal-level reporting subscriptions + on the managed path (see :mod:`adcp.reporting.subscriptions.managed`): + its presence enables them, and the service derives the principal + runtime, roster, buys, offerings and scheduling from the platform it is + installed on. ``reporting_subscriptions`` is the advanced composition, + where the host builds and links the engine itself. Without either the + service is unchanged.""" if production is not None and ( caller_resolver is not None or producer_factory is not ReportingProducer @@ -516,6 +543,10 @@ def __init__( raise ReliableReportingConfigurationError( "production options compose their own workers, admission and receipt handlers" ) + if subscriptions is not None and reporting_subscriptions is not None: + raise ReliableReportingConfigurationError( + "pass subscriptions (managed) or reporting_subscriptions (advanced), not both" + ) if ( production is not None and reporting_subscriptions is not None @@ -525,6 +556,15 @@ def __init__( "production serves reporting subscriptions through managed delivery; " "supply reporting_subscriptions.bind_managed_generation" ) + if ( + production is not None + and subscriptions is not None + and subscriptions.bind_managed_generation is None + ): + raise ReliableReportingConfigurationError( + "production serves reporting subscriptions through managed delivery; " + "supply SubscriptionOptions.bind_managed_generation" + ) effective_clock = clock or (lambda: datetime.now(timezone.utc)) self.store = store self._production: ReportingProductionSupport | None = None @@ -571,21 +611,21 @@ def __init__( self._turn_lock = asyncio.Lock() self._subscriptions = reporting_subscriptions self._subscription_planner: Any | None = None - if reporting_subscriptions is not None and production is None: - from adcp.reporting.subscriptions.ledger import SubscriptionObligationPlanner + self._managed: ManagedSubscriptions | None = None + if subscriptions is not None: + from adcp.reporting.subscriptions.managed import ManagedSubscriptions try: - self._subscription_planner = SubscriptionObligationPlanner( - store, - offerings=reporting_subscriptions.engine.offerings, - currency=self._subscription_currency, - definitions=reporting_subscriptions.definitions, - automated_recovery_window=automated_recovery_window, + self._managed = ManagedSubscriptions( + subscriptions, + clock=effective_clock, status_retention_days=status_retention_days, - install=self._install_subscription_generation, + report=self._report_planning_failure, ) except TypeError as error: raise ReliableReportingConfigurationError(str(error)) from None + if reporting_subscriptions is not None and production is None: + self._build_subscription_planner() self._lifecycle = _ServiceLifecycle( self._initialize, resources=( @@ -665,6 +705,97 @@ def memory( **kwargs, ) + def _build_subscription_planner(self) -> None: + from adcp.reporting.subscriptions.ledger import SubscriptionObligationPlanner + + assert self._subscriptions is not None + try: + self._subscription_planner = SubscriptionObligationPlanner( + self.store, + offerings=self._subscriptions.engine.offerings, + currency=self._subscription_currency, + definitions=self._subscriptions.definitions, + automated_recovery_window=self._automated_recovery_window, + status_retention_days=self._status_retention_days, + install=self._install_subscription_generation, + ) + except TypeError as error: + raise ReliableReportingConfigurationError(str(error)) from None + + def _compose_managed(self) -> None: + """Build the managed subscription runtime from the registered offerings (once).""" + managed = self._managed + if managed is None or managed.composed: + return + if self._production_options is not None: + offerings = [_wire(item.offering) for item in self._production_options.offerings] + else: + offerings = [ + dict(item) + for name in self.sources.names + for item in self.sources.get(name).capability_offerings + ] + self._subscriptions = managed.compose([_thaw(item) for item in offerings]) + if self._production_options is None: + self._build_subscription_planner() + + @property + def principals(self) -> Any: + """The principal runtime the managed subscription path owns. + + A :class:`~adcp.server.PrincipalService`; use it, for example, to make + a ``sync_accounts`` implementation reject a ``delivery_config_id`` that + equals one of the caller's subscription ids. + """ + if self._managed is None: + raise ReliableReportingConfigurationError( + "the principal runtime belongs to ReliableReportingService(subscriptions=...)" + ) + self._compose_managed() + return self._managed.principals + + async def get_principal(self, request: Any, context: Any | None = None) -> Any: + """Serve ``get_principal`` from the managed principal runtime.""" + managed = self._require_managed() + return await managed.get_principal(request, context) + + async def sync_principal(self, request: Any, context: Any | None = None) -> Any: + """Serve ``sync_principal`` and reconcile the principal when it changed.""" + managed = self._require_managed() + return await managed.sync_principal(request, context) + + async def touch(self, *, account_id: str | None = None, context: Any | None = None) -> None: + """Reconcile now after a change the SDK did not serve (a seller UI, a batch job). + + Reconciles every principal whose subscriptions include ``account_id`` + (and the principal ``context`` authenticates, if given). The worker's + boundary sweep catches such changes anyway; this is for immediacy. + """ + managed = self._require_managed() + await managed.touch(account_id=account_id, context=context) + + def _require_managed(self) -> ManagedSubscriptions: + if self._managed is None: + raise ReliableReportingConfigurationError( + "principal tasks are served by ReliableReportingService(subscriptions=...)" + ) + self._compose_managed() + return self._managed + + def _bind_handler(self, handler: Any) -> None: + """Called by the decisioning ``PlatformHandler`` serving the installed platform.""" + if self._managed is not None: + self._managed.bind_handler(handler) + + async def _subscription_write(self, account_id: str | None, context: Any | None) -> None: + """Trigger: a served account or media-buy write may change what is owed.""" + if self._managed is None or not self._managed.composed: + return + try: + await self._managed.touch(account_id=account_id, context=context) + except Exception: # noqa: BLE001 - a trigger never fails the buyer's write + logger.exception("Reliable Reporting subscription trigger failed") + def _compose_production(self) -> None: if self._production_options is None or self._production is not None: return @@ -816,6 +947,9 @@ async def _production_subscription_cycle(self) -> None: assert self._production is not None and self._subscriptions is not None production = self._production + if self._managed is not None: + for failure in await self._managed.sweep(now=self._clock()): + await self._report_planning_failure(failure) _, failures = await plan_subscription_cycle( self._subscriptions, self._subscription_planner, @@ -859,8 +993,11 @@ async def _configure( f"for account {key.account_id!r} is already frozen with other facts" ) return + resolver = self._context_resolver + if configuration.owed_set and self._managed is not None: + resolver = self._managed.options.configure or resolver try: - context = await _resolve(self._context_resolver(configuration)) + context = await _resolve(resolver(configuration)) except Exception as error: if recovering: raise _AccountContextResolutionError(error) from error @@ -1091,6 +1228,7 @@ def _validate_offerings( def validate(self) -> None: """Fail fast on tier combinations the configured components cannot honor.""" + self._compose_managed() self._compose_production() if self._production is not None and self._production_options is None and self.sources.names: raise ReliableReportingConfigurationError( @@ -1296,8 +1434,15 @@ async def _run_worker(self, *, now: datetime | None) -> ReliableReportingTurn: await self._recover_configuration(configuration, retrying=True) turn.configuration_errors.update(self._initialization_errors) if self._subscriptions is not None and not self._lifecycle.stopping: + instant = now or self._clock() + if self._managed is not None: + # The boundary sweep: freeze every period that ended since + # its principal's last reconcile before planning it. + for failure in await self._managed.sweep(now=instant): + turn.subscription_errors[failure.key] = failure.error + await self._report_planning_failure(failure) # Plan first, so newly frozen obligations run in this turn. - await self._plan_subscriptions(turn, now=now or self._clock()) + await self._plan_subscriptions(turn, now=instant) with source_turn(): for key, binding in sorted( self._bindings.items(), @@ -1528,6 +1673,7 @@ def capability_block(self, *, adcp_version: str | None = None) -> dict[str, Any] ``adcp_version`` is the release the response is served at; it decides which reporting-subscription fields appear when subscriptions are served. """ + self._compose_managed() block = self._capability_block() if self._subscriptions is None or not block: return block @@ -1616,8 +1762,13 @@ def inject_capabilities( experimental.append("media_buy.reporting_delivery") payload["experimental_features"] = experimental if self._subscriptions is not None: - from adcp.reporting.service_subscriptions import advertise_subscriptions + from adcp.reporting.service_subscriptions import ( + advertise_subscriptions, + serves_subscriptions, + ) + if self._managed is not None and serves_subscriptions(adcp_version): + payload = self._managed.advertise_principal(payload) try: payload = advertise_subscriptions( payload, self._subscriptions, adcp_version=adcp_version @@ -1635,14 +1786,23 @@ def install(self, platform: Any) -> Any: bind an ADCPHandler and return their exact production handler; use ``create_adcp_server_from_platform`` first for that composition. """ + self._compose_managed() self._compose_production() if self._production is not None: + if self._managed is not None: + # The production handler delegates ordinary tasks to the + # application, so the principal tasks and write triggers + # are installed there. + platform = self._install_principal_tasks(platform) + self._managed.bind_platform(platform, decisioning=False) self._production.handler.bind_application(platform) return self._production.handler from adcp.server import ADCPHandler from adcp.server.mcp_tools import get_tools_for_handler decisioning = not isinstance(platform, ADCPHandler) + if self._managed is not None: + self._managed.bind_platform(platform, decisioning=decisioning) if decisioning: from adcp.decisioning.platform import DecisioningPlatform @@ -1735,15 +1895,18 @@ async def get_media_buy_delivery( get_adcp_capabilities=get_adcp_capabilities, get_media_buy_delivery=get_media_buy_delivery, ) + if self._managed is not None: + attrs.update(self._principal_task_methods(original)) reporting_methods = { "get_adcp_capabilities", "get_reporting_status", "sync_reporting_status", "sync_reporting_receipts", "get_media_buy_delivery", + *(_MANAGED_TASKS if self._managed is not None and not decisioning else ()), } for name in existing_tools: - if name in reporting_methods: + if name in reporting_methods or name in attrs: continue method = getattr(original, name, None) if method is None: @@ -1785,10 +1948,69 @@ async def delegated(instance: Any, *args: Any, **kwargs: Any) -> Any: setattr(platform, "_reliable_reporting_service", self) return platform + def _principal_task_methods(self, original: type) -> dict[str, Any]: + """``get_principal``/``sync_principal`` and the write triggers for an ADCPHandler.""" + service = self + + async def get_principal(instance: Any, params: Any, context: Any | None = None) -> Any: + return await installed_reporting_call( + "get_principal", lambda: service.get_principal(params, context) + ) + + async def sync_principal(instance: Any, params: Any, context: Any | None = None) -> Any: + return await installed_reporting_call( + "sync_principal", lambda: service.sync_principal(params, context) + ) + + def triggered(name: str) -> Callable[..., Any] | None: + method = getattr(original, name, None) + if method is None: + return 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) + return result + + write.__name__ = name + write.__qualname__ = f"ReliableReporting{original.__name__}.{name}" + return write + + methods: dict[str, Any] = { + "get_principal": get_principal, + "sync_principal": sync_principal, + } + for name in ("sync_accounts", "create_media_buy", "update_media_buy"): + wrapped = triggered(name) + if wrapped is not None: + methods[name] = wrapped + return methods + + def _install_principal_tasks(self, application: Any) -> Any: + """Install the managed principal tasks on a production application handler.""" + from adcp.server import ADCPHandler + + if not isinstance(application, ADCPHandler): + raise ReliableReportingConfigurationError( + "production installs bind an ADCPHandler application" + ) + original = type(application) + attrs = {"__module__": original.__module__, **self._principal_task_methods(original)} + installed = type(f"ReliableReportingPrincipal{original.__name__}", (original,), attrs) + try: + application.__class__ = installed + except TypeError: + raise ReliableReportingConfigurationError( + "install() requires an ordinary mutable application instance" + ) from None + return application + @property def reporting_tools(self) -> frozenset[str]: """Core task implementations mounted by ``install`` (independent of readiness).""" tasks = {"get_reporting_status", "get_media_buy_delivery"} + if self._managed is not None: + tasks.update(_MANAGED_TASKS) if self._consumer_status_enabled: tasks.add("sync_reporting_status") if self._receipt_handler is not None: diff --git a/src/adcp/reporting/service_subscriptions.py b/src/adcp/reporting/service_subscriptions.py index ab5273b67..6157f61d6 100644 --- a/src/adcp/reporting/service_subscriptions.py +++ b/src/adcp/reporting/service_subscriptions.py @@ -1,7 +1,12 @@ """Reporting subscriptions in :class:`~adcp.reporting.service.ReliableReportingService`. -Opt in with ``ReliableReportingService(reporting_subscriptions=...)``. Without -it, capabilities, responses and worker turns are exactly what they were. +Most sellers opt in with ``ReliableReportingService(subscriptions=SubscriptionOptions(...))``, +the managed path (:mod:`adcp.reporting.subscriptions.managed`), which builds +everything below from the platform the service is installed on. +``ReliableReportingService(reporting_subscriptions=ReportingSubscriptionServiceOptions(...))`` +is the advanced composition, where the host builds the engine and links the +principal runtime itself. Without either, capabilities, responses and worker +turns are exactly what they were. With it, the service: @@ -39,9 +44,11 @@ from datetime import datetime from typing import TYPE_CHECKING, Any +from adcp._principal_ref import PrincipalRef from adcp._version import is_adcp_version_at_least, resolve_adcp_version_alias from adcp.reporting._errors import ReliableReportingConfigurationError from adcp.reporting.ledger.models import ReportingConfiguration, ReportingDefinitionBinding +from adcp.reporting.subscriptions.managed import SubscriptionOptions if TYPE_CHECKING: from adcp.reporting.ledger.delivery_models import ( @@ -53,19 +60,23 @@ #: Every principal whose subscription obligations the service writes. PrincipalLister = Callable[ - [], "Sequence[PrincipalIdentity] | Awaitable[Sequence[PrincipalIdentity]]" + [], + "Sequence[PrincipalIdentity | PrincipalRef] | Awaitable[Sequence[PrincipalIdentity | PrincipalRef]]", # noqa: E501 ] #: The authenticated principal of a request; never derived from request content. -PrincipalResolver = Callable[[Any, Any], "PrincipalIdentity | Awaitable[PrincipalIdentity]"] +PrincipalResolver = Callable[ + [Any, Any], + "PrincipalIdentity | PrincipalRef | Awaitable[PrincipalIdentity | PrincipalRef]", +] _FIRST_RELEASE = "3.3.0-beta.1" @dataclass(frozen=True) class SubscriptionPlanningFailure: - """One subscription (or one principal's listing) a turn could not plan.""" + """One subscription (or one principal's listing or reconcile) a turn could not plan.""" - principal: PrincipalIdentity + principal: PrincipalIdentity | PrincipalRef error: Exception account_id: str | None = None subscription_id: str | None = None @@ -353,6 +364,7 @@ def check_managed_binding( "ManagedGenerationBinder", "ProductionSubscriptionHooks", "ReportingSubscriptionServiceOptions", + "SubscriptionOptions", "SubscriptionPlanningFailure", "serves_subscriptions", ] diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index 79445219c..e7d918da3 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -42,6 +42,7 @@ from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Any, Protocol, TypeVar +from adcp._principal_ref import PrincipalRef from adcp.reporting.subscriptions._clock import PeriodCalendar, iso_z, require_zone, utc_instant from adcp.reporting.subscriptions.capabilities import ReportingSubscriptionSupport from adcp.reporting.subscriptions.coverage import ( @@ -83,6 +84,10 @@ ReportingSubscriptionOptions, ) + #: Either spelling of a principal: reporting APIs take :class:`PrincipalRef`; + #: a :class:`~adcp.server.PrincipalIdentity` is still accepted. + PrincipalLike = PrincipalRef | PrincipalIdentity + # -- host facts ---------------------------------------------------------------- @@ -150,11 +155,11 @@ class ReportingSubscriptionHost(Protocol): """Facts only the seller's host knows. The SDK never infers them.""" async def list_authorized_accounts( - self, principal: PrincipalIdentity, *, cursor: str | None + self, principal: PrincipalLike, *, cursor: str | None ) -> AuthorizedAccountPage: ... async def account_buys( - self, principal: PrincipalIdentity, account_id: str + self, principal: PrincipalLike, account_id: str ) -> Sequence[SubscriptionBuy]: ... @@ -169,7 +174,7 @@ class ReportingSubscriptionDeliveryHost(Protocol): async def is_period_delivered( self, - principal: PrincipalIdentity, + principal: PrincipalLike, account_id: str, period: ExpectedSubscriptionPeriod, *, @@ -181,10 +186,31 @@ class ReportingSubscriptionPrecedenceHost(Protocol): """Optional: the caller's own account-level configurations, for precedence.""" async def account_delivery_configs( - self, principal: PrincipalIdentity, account_id: str + self, principal: PrincipalLike, account_id: str ) -> Sequence[AccountDeliveryConfig]: ... +class AccountConfigsEnder(Protocol): + """Ends the caller's own account-level configurations for ``replace_account_configs``. + + Called with the principal, the account, the ``delivery_config_id`` values + the host's ``account_delivery_configs`` lists for the subscription's + ``feed_purpose``, and the instant the subscription takes over. The host + deactivates exactly those configurations, never another principal's, and + never advances the account's ``revision``. It must be idempotent: a failed + reconcile retries it. + """ + + async def __call__( + self, + principal: PrincipalLike, + account_id: str, + delivery_config_ids: tuple[str, ...], + *, + at: datetime, + ) -> None: ... + + # -- engine state -------------------------------------------------------------- @@ -262,6 +288,16 @@ class _PrincipalState: #: (subscription_id, account_id, unauthorized_at) -> withheld count from the #: host's delivery facts; delivery by a past instant never changes withheld: dict[tuple[str, str, datetime], int] = field(default_factory=dict) + #: The principal's last authenticated binding (tenant, kind and a + #: token-free auth snapshot), so the managed path can read the host's + #: accounts and buys under the principal's identity outside a request. + binding: dict[str, Any] | None = None + #: When the principal was last reconciled; the worker's boundary sweep + #: reconciles it again once a period boundary passes after this. + reconciled_at: datetime | None = None + #: subscription_id -> (version, replace_account_configs) last applied, so + #: ``replace_account_configs`` runs once per version or on a change to true. + replacements: dict[str, tuple[int, bool]] = field(default_factory=dict) #: Version of the persisted engine-state document; bump on a breaking change. @@ -307,6 +343,10 @@ async def compare_and_swap( """ ... + # Optional: ``async def list_principals(self) -> Sequence[str]`` returns + # every subject with stored state. The managed path needs it to find the + # principals each worker turn plans; both SDK stores implement it. + class InMemoryReportingSubscriptionStore: """Process-local :class:`ReportingSubscriptionStateStore` for tests and single workers. @@ -341,6 +381,11 @@ async def compare_and_swap( ) return True + async def list_principals(self) -> list[str]: + """Every subject with stored state, in subject order.""" + async with self._lock: + return sorted(self._documents) + class ReportingSubscriptionStateConflictError(RuntimeError): """The engine could not write its state after repeated concurrent changes.""" @@ -466,6 +511,24 @@ def encode_principal_state( } for (sid, account_id, lost), count in sorted(principal.withheld.items()) ], + # Present only once set, so documents written before these existed + # stay byte-identical. + **({"binding": _json_value(principal.binding)} if principal.binding is not None else {}), + **( + {"reconciled_at": iso_z(principal.reconciled_at)} + if principal.reconciled_at is not None + else {} + ), + **( + { + "replacements": { + sid: {"version": version, "replace_account_configs": flag} + for sid, (version, flag) in sorted(principal.replacements.items()) + } + } + if principal.replacements + else {} + ), } @@ -533,6 +596,11 @@ def decode_principal_state(document: Mapping[str, Any]) -> _PrincipalState: _required_instant(item["unauthorized_at"]), ) ] = int(item["count"]) + if document.get("binding") is not None: + principal.binding = dict(document["binding"]) + principal.reconciled_at = _instant(document.get("reconciled_at")) + for sid, item in (document.get("replacements") or {}).items(): + principal.replacements[sid] = (int(item["version"]), bool(item["replace_account_configs"])) return principal @@ -561,7 +629,12 @@ def __init__( status_retention_days: int = 400, store: ReportingSubscriptionStateStore | None = None, clock: Callable[[], datetime] | None = None, + end_account_configs: AccountConfigsEnder | None = None, ) -> None: + """``end_account_configs`` enables ``replace_account_configs``: without it + the principal runtime refuses the flag with ``UNSUPPORTED_FEATURE``. With + it, the engine applies the flag once per ``subscription_version``, or when + it changes to true, through the host's ``account_delivery_configs``.""" ids = [item.offering_id for item in offerings] if len(set(ids)) != len(ids): raise ValueError("offering_id values must be unique") @@ -586,6 +659,14 @@ def __init__( self._store = store or InMemoryReportingSubscriptionStore() self._clock = clock or (lambda: datetime.now(timezone.utc)) self._errors: dict[tuple[str, str], Exception] = {} + if end_account_configs is not None and not callable( + getattr(host, "account_delivery_configs", None) + ): + raise ValueError( + "end_account_configs needs the host's account_delivery_configs, which names " + "the configurations replace_account_configs ends" + ) + self._end_account_configs = end_account_configs @property def evaluation_errors(self) -> Mapping[tuple[str, str], Exception]: @@ -619,8 +700,86 @@ def principal_options( delivery_config_ids=delivery_config_ids, record_versions=self.record_versions, clock=self._clock, + replace_account_configs=self._end_account_configs is not None, ) + @property + def store(self) -> ReportingSubscriptionStateStore: + """The engine-state store every worker shares.""" + return self._store + + async def list_principals(self) -> list[PrincipalRef]: + """Every principal the store holds engine state for, in subject order. + + Requires a store that implements ``list_principals()`` (both SDK stores + do); each principal's tenant comes from its recorded binding. + """ + lister = getattr(self._store, "list_principals", None) + if not callable(lister): + raise TypeError( + "the reporting-subscription state store does not implement list_principals(); " + "supply list_principals to the service instead" + ) + result: list[PrincipalRef] = [] + for subject in sorted(await lister()): + binding = (await self._read(subject)).binding or {} + result.append(PrincipalRef(binding.get("tenant_id"), subject)) + return result + + async def record_binding(self, principal: PrincipalLike, binding: Mapping[str, Any]) -> None: + """Remember the principal's last authenticated binding (no credentials). + + The managed path reads the host's accounts and buys under it outside a + request. ``binding`` is plain JSON; a principal recorded under another + tenant is refused, because one engine serves one tenant. + """ + document = _json_value(dict(binding)) + tenant = document.get("tenant_id") + + def change(state: _PrincipalState) -> None: + if state.binding is not None and state.binding.get("tenant_id") != tenant: + raise ValueError( + f"principal {principal.subject!r} is recorded under another tenant" + ) + state.binding = document + + await self._mutate(principal.subject, change) + + 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) + + async def binding(self, principal: PrincipalLike) -> Mapping[str, Any] | None: + """The binding :meth:`record_binding` last stored, if any.""" + return (await self._read(principal.subject)).binding + + async def boundary_due(self, principal: PrincipalLike, *, now: datetime | None = None) -> bool: + """Whether a period boundary passed since the principal was last reconciled. + + True for a principal never reconciled. Checks every offering's calendar + in every zone the principal's accounts use, so a period is frozen at + the first reconcile after it ends. + """ + instant = self._now(now) + state = await self._read(principal.subject) + last = state.reconciled_at + if last is None: + return True + if instant <= last: + return False + zones = {account.timezone for account in state.accounts.values()} or {None} + for offering in self._offerings: + for zone in zones: + if offering.alignment == "account_timezone" and zone is None: + continue + try: + calendar = offering.calendar(zone) + except Exception: # noqa: BLE001 - an unusable zone is reported by reconcile + return True + if calendar.floor(instant) > last: + return True + return False + def _now(self, now: datetime | None) -> datetime: return utc_instant(now if now is not None else self._clock(), field="now") @@ -662,7 +821,7 @@ async def _mutate( # -- recording ------------------------------------------------------------- async def record_versions( - self, identity: PrincipalIdentity, states: Sequence[ReportingSubscriptionState] + self, identity: PrincipalLike, states: Sequence[ReportingSubscriptionState] ) -> None: """Keep every version's configuration; the readback keeps only the current one. @@ -689,7 +848,7 @@ def change(principal: _PrincipalState) -> None: async def authorization_changed( self, - identity: PrincipalIdentity, + identity: PrincipalLike, account_id: str, *, authorized: bool, @@ -729,7 +888,7 @@ def _apply_grant(account: _Account, *, authorized: bool, at: datetime) -> None: last.authorized_at, max(at, last.authorized_at + timedelta(microseconds=1)) ) - async def _list_accounts(self, identity: PrincipalIdentity) -> list[AuthorizedAccount]: + async def _list_accounts(self, identity: PrincipalLike) -> list[AuthorizedAccount]: accounts: list[AuthorizedAccount] = [] cursor: str | None = None while True: @@ -741,7 +900,7 @@ async def _list_accounts(self, identity: PrincipalIdentity) -> list[AuthorizedAc # -- reconcile ------------------------------------------------------------- - async def reconcile(self, identity: PrincipalIdentity, *, now: datetime | None = None) -> None: + async def reconcile(self, identity: PrincipalLike, *, now: datetime | None = None) -> None: """Diff the host's authorized accounts, then re-evaluate every exception. Call it on a schedule and whenever the host knows something changed @@ -787,11 +946,83 @@ def change(principal: _PrincipalState) -> None: self._apply_grant(account, authorized=False, at=instant) if account_id not in listed: account.allowed_tasks = None + principal.reconciled_at = max(instant, principal.reconciled_at or instant) await self._mutate(identity.subject, change, now=instant) + await self._apply_replacements(identity, instant) await self._evaluate(identity, instant) - async def _evaluate(self, identity: PrincipalIdentity, now: datetime) -> None: + async def _apply_replacements(self, identity: PrincipalLike, now: datetime) -> None: + """Apply ``replace_account_configs`` once per version, or on a change to true. + + Ends the caller's own account-level configurations with the + subscription's ``feed_purpose`` on every authorized account the scope + includes where the selector resolves an offering. Marks are recorded + only after the host ends them, so a failure is retried by the next + reconcile. + """ + states = [_dump(item) for item in await self._principals.reporting_subscriptions(identity)] + principal = await self._read(identity.subject) + marks: dict[str, tuple[int, bool]] = {} + pending: list[Mapping[str, Any]] = [] + for state in states: + subscription_id = str(state["subscription_id"]) + version = int(state["subscription_version"]) + flag = _active(state) and state["configuration"].get("replace_account_configs") is True + if principal.replacements.get(subscription_id) == (version, flag): + continue + marks[subscription_id] = (version, flag) + if flag: + pending.append(state) + if not marks: + return + if pending: + ender = self._end_account_configs + if ender is None: + # Accepted only when the host can apply it; the principal + # runtime refuses the flag otherwise. + raise ValueError("replace_account_configs needs end_account_configs") + listing = getattr(self._host, "account_delivery_configs") + destinations = await self._destinations(identity) + for replacing in pending: + configuration = replacing["configuration"] + feed_purpose = configuration["selector"]["feed_purpose"] + selector = SubscriptionSelector.from_subscription(configuration) + destination = _destination_for(selector, destinations) + for account_id, account in sorted(principal.accounts.items()): + if not account.authorized or not _scope_selects( + configuration, account_id, account + ): + continue + buys = await self._host.account_buys(identity, account_id) + product_ids = _buy_offering_ids(buys) or [ + item.offering_id for item in self._offerings + ] + resolution = resolve_subscription_offerings( + selector, + self._offerings, + product_offering_ids=product_ids, + destination=destination, + at=now, + ) + if not resolution.offerings: + continue + ids = tuple( + sorted( + config.delivery_config_id + for config in await listing(identity, account_id) + if config.feed_purpose == feed_purpose + ) + ) + if ids: + await ender(identity, account_id, ids, at=now) + + def record(state: _PrincipalState) -> None: + state.replacements.update(marks) + + await self._mutate(identity.subject, record) + + async def _evaluate(self, identity: PrincipalLike, now: datetime) -> None: states = [_dump(item) for item in await self._principals.reporting_subscriptions(identity)] destinations = await self._destinations(identity) precedence = getattr(self._host, "account_delivery_configs", None) @@ -927,7 +1158,7 @@ def _evaluate_subscription( self._accounts_unauthorized(principal, state, accounts, buys, destinations, now), ) - async def _destinations(self, identity: PrincipalIdentity) -> dict[str, Any]: + async def _destinations(self, identity: PrincipalLike) -> dict[str, Any]: return { item.destination_id: item.configuration for item in await self._principals.reporting_destinations(identity) @@ -1197,7 +1428,7 @@ def _accounts_unauthorized( async def subscription_coverage( self, - identity: PrincipalIdentity, + identity: PrincipalLike, account_id: str, *, now: datetime | None = None, @@ -1375,7 +1606,7 @@ def cutover(at: datetime, _same: datetime) -> dict[str, Any]: async def owed_periods( self, - identity: PrincipalIdentity, + identity: PrincipalLike, subscription_id: str, account_id: str, *, @@ -1458,7 +1689,7 @@ def _owed( async def frozen_owed_sets( self, - identity: PrincipalIdentity, + identity: PrincipalLike, *, account_id: str | None = None, now: datetime | None = None, @@ -1511,7 +1742,7 @@ async def frozen_owed_sets( ) return result - async def grants_task(self, identity: PrincipalIdentity, account_id: str, task: str) -> bool: + 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 @@ -1520,17 +1751,17 @@ async def grants_task(self, identity: PrincipalIdentity, account_id: str, task: account = (await self._read(identity.subject)).accounts.get(account_id) return account is not None and account.grants_task(task) - async def authorizes_delivery(self, subject: str, account_id: str) -> bool: + async def authorizes_delivery(self, principal: PrincipalLike, account_id: str) -> bool: """Whether a principal's current grant still carries its subscriptions on the account. For managed delivery's per-attempt authority check (``ReportingMaterializerService.authorize_subscription_delivery``): the grant must list ``get_reporting_status``, as coverage requires. """ - account = (await self._read(subject)).accounts.get(account_id) + account = (await self._read(principal.subject)).accounts.get(account_id) return account is not None and account.grants_task(REPORTING_STATUS_TASK) - async def receipt_roster(self, identity: PrincipalIdentity, account_id: str) -> list[str]: + async def receipt_roster(self, identity: PrincipalLike, account_id: str) -> list[str]: """Active ``consumer_receipt`` subscriptions covering the account. A consumer owes receipts for an account while one of its active diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py new file mode 100644 index 000000000..69df09bde --- /dev/null +++ b/src/adcp/reporting/subscriptions/managed.py @@ -0,0 +1,890 @@ +"""The managed seller path for reporting subscriptions (AdCP 3.3, experimental). + +``ReliableReportingService(subscriptions=SubscriptionOptions(...))`` enables +``media_buy.reporting_subscriptions``. The block's presence is the switch; the +SDK derives everything else from surfaces the seller already serves: + +* **Principal runtime.** The service owns a + :class:`~adcp.server.PrincipalService`, links it to the engine, and + ``install()`` serves ``get_principal`` and ``sync_principal`` from it and + advertises the principal section. +* **Roster.** The accounts a principal may read come from the seller's own + ``list_accounts`` (a decisioning platform's ``accounts.list``, or an + ``ADCPHandler.list_accounts``), invoked under the principal's identity: a + service context carrying the principal's last authenticated binding, with + no credential material. +* **Buys.** Each account's buys come from the seller's ``get_media_buys`` + under the same identity, so visibility matches what the principal can see. +* **Offerings.** From the service's registered capability offerings, each + of which must publish ``standard_dimensions``. +* **Scheduling.** ``sync_principal``, ``sync_accounts``, ``create_media_buy`` + and ``update_media_buy`` served through the installed platform reconcile the + principals concerned, and every worker turn reconciles each principal a + period boundary has passed for before it plans obligations. + +Each derivation has an override on :class:`SubscriptionOptions`. A seller +whose accounts are only reachable through a live upstream token supplies +``accounts``: a listing that fails under the service context raises +``SUBSCRIPTION_ROSTER_UNAVAILABLE`` and never reads as an empty roster. +""" + +from __future__ import annotations + +import inspect +import logging +from collections.abc import Awaitable, Callable, Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime +from typing import TYPE_CHECKING, Any + +from adcp._principal_ref import PrincipalRef +from adcp.reporting._errors import ReliableReportingConfigurationError +from adcp.reporting.subscriptions._clock import utc_instant +from adcp.reporting.subscriptions.coverage import SubscriptionBuy, SubscriptionBuyPackage +from adcp.reporting.subscriptions.engine import ( + AccountConfigsEnder, + AccountDeliveryConfig, + AuthorizedAccount, + AuthorizedAccountPage, + ReportingSubscriptionStateStore, +) + +if TYPE_CHECKING: + from adcp.reporting.ledger.models import ReportingDefinitionBinding + from adcp.reporting.service_subscriptions import ( + ManagedGenerationBinder, + PlanningErrorHandler, + ReportingSubscriptionServiceOptions, + SubscriptionPlanningFailure, + ) + from adcp.reporting.subscriptions.coverage import ExpectedSubscriptionPeriod + from adcp.reporting.subscriptions.engine import ReportingSubscriptionEngine + from adcp.reporting.subscriptions.resolution import SubscriptionOffering + from adcp.server.principal import ( + PrincipalChangeEmitter, + PrincipalRecordStore, + PrincipalService, + ) + +logger = logging.getLogger(__name__) + +#: ``(principal, *, cursor)`` -> one page of the accounts the principal may read. +AccountsOverride = Callable[..., "Awaitable[AuthorizedAccountPage] | AuthorizedAccountPage"] +#: ``(principal, account_id)`` -> the buys the principal can see on the account. +BuysOverride = Callable[ + [PrincipalRef, str], "Awaitable[Sequence[SubscriptionBuy]] | Sequence[SubscriptionBuy]" +] +#: ``(media_buy)`` -> each package's ``reporting_delivery_offering_ids``, keyed by +#: ``package_id``; ``media_buy`` is the wire ``get_media_buys`` entry. +BuyOfferingsOverride = Callable[[Mapping[str, Any]], Mapping[str, Sequence[str]]] +#: ``(principal, account_id)`` -> the caller's own active account-level configurations. +AccountConfigsOverride = Callable[ + [PrincipalRef, str], + "Awaitable[Sequence[AccountDeliveryConfig]] | Sequence[AccountDeliveryConfig]", +] +#: ``(principal, account_id, period, *, by)`` -> whether the period was delivered by ``by``. +PeriodDeliveredOverride = Callable[..., "Awaitable[bool] | bool"] +#: ``(request, context)`` -> the authenticated principal, never derived from request content. +ManagedPrincipalResolver = Callable[[Any, Any], Any] + +#: The SDK's request-context key for the verified :class:`~adcp.decisioning.AuthInfo`. +_AUTH_INFO_KEY = "adcp.auth_info" +_ALL_BUY_STATUSES = ( + "pending_creatives", + "pending_start", + "active", + "paused", + "completed", + "rejected", + "canceled", +) +#: Hard cap on pages walked for one principal's roster or one account's buys; +#: a walk that does not end is refused, never truncated. +_MAX_PAGES = 10_000 + + +@dataclass(frozen=True) +class SubscriptionOptions: + """Serve ``media_buy.reporting_subscriptions`` from ``ReliableReportingService``. + + Its presence enables the feature. ``max_accounts`` and + ``backfill_on_regrant`` are advertised as ``subscriptions.max_accounts`` + and ``subscriptions.backfill_on_regrant``; ``dimensions`` defaults to the + union of the offerings' ``standard_dimensions``. + + Every other field is an optional override of something the SDK derives: + + * ``accounts(principal, *, cursor)``: the roster, instead of the seller's + ``list_accounts`` under the principal's identity. Required when account + access depends on a live upstream token. + * ``buys(principal, account_id)``: an account's buys, instead of + ``get_media_buys``. + * ``buy_offerings(media_buy)``: each package's offering ids, keyed by + ``package_id``. Without it every package is reported by every offering + the service registered; supply it when products differ. + * ``configure(generation)``: the report content for a subscription + generation, instead of the service's ``account_context``. + * ``account_delivery_configs(principal, account_id)``: the caller's own + active ``sync_accounts`` configurations, so they take precedence. + * ``end_account_configs(principal, account_id, ids, *, at)``: ends them for + ``replace_account_configs`` (needs ``account_delivery_configs``). Without + it the flag is refused with ``UNSUPPORTED_FEATURE``. + * ``is_period_delivered(principal, account_id, period, *, by)``: exact + ``withheld_period_count``. + * ``resolve_principal(request, context)``: the authenticated principal. + By default it is the caller the reporting tasks resolve (the bearer + principal, the verified signed-request principal, or + ``ToolContext.caller_identity``), with ``ToolContext.tenant_id``. + + ``reconcile_on_write`` (default on) reconciles a principal when a served + write changes what it is owed. ``principals``, ``principal_store`` and + ``state_store`` supply the principal runtime and the stores; share durable + stores across every worker in production. ``account_configuration`` + keeps ``sync_accounts`` in ``configuration_tasks``. + """ + + max_accounts: int | None = None + backfill_on_regrant: bool = False + dimensions: Sequence[str] | None = None + accounts: AccountsOverride | None = None + buys: BuysOverride | None = None + buy_offerings: BuyOfferingsOverride | None = None + configure: Callable[[Any], Any] | None = None + account_delivery_configs: AccountConfigsOverride | None = None + end_account_configs: AccountConfigsEnder | None = None + is_period_delivered: PeriodDeliveredOverride | None = None + resolve_principal: ManagedPrincipalResolver | None = None + reconcile_on_write: bool = True + account_configuration: bool = True + principal_kind: str = "buyer_agent" + principals: PrincipalService | None = None + principal_store: PrincipalRecordStore | None = None + state_store: ReportingSubscriptionStateStore | None = None + emit_change: PrincipalChangeEmitter | None = None + definitions: Mapping[str, ReportingDefinitionBinding] | None = None + bind_managed_generation: ManagedGenerationBinder | None = None + planning_error_handler: PlanningErrorHandler | None = None + + def __post_init__(self) -> None: + if self.max_accounts is not None and ( + type(self.max_accounts) is not int or self.max_accounts < 1 + ): + raise ValueError("SubscriptionOptions.max_accounts must be a positive integer") + if self.principals is not None and self.principal_store is not None: + raise ValueError("pass principals or principal_store, not both") + if self.end_account_configs is not None and self.account_delivery_configs is None: + raise ValueError( + "SubscriptionOptions.end_account_configs needs account_delivery_configs, which " + "names the configurations replace_account_configs ends" + ) + for name in ( + "accounts", + "buys", + "buy_offerings", + "configure", + "account_delivery_configs", + "end_account_configs", + "is_period_delivered", + "resolve_principal", + ): + value = getattr(self, name) + if value is not None and not callable(value): + raise TypeError(f"SubscriptionOptions.{name} must be callable") + + +async def _resolve(value: Any) -> Any: + return await value if inspect.isawaitable(value) else value + + +def _wire(value: Any) -> Any: + if hasattr(value, "model_dump"): + return value.model_dump(mode="json", exclude_none=True) + return value + + +def _instant(value: Any) -> datetime | None: + if value is None: + return None + if isinstance(value, datetime): + return utc_instant(value) + return utc_instant(datetime.fromisoformat(str(value).replace("Z", "+00:00"))) + + +# -- principal identity -------------------------------------------------------- + + +def request_principal_id(context: Any) -> str | None: + """The authenticated caller a request context carries, as the reporting tasks read it. + + The same derivation as ``ReliableReportingService``'s default caller: an + ``AccountAwareToolContext`` (or a plain ``ToolContext``) contributes its + ``caller_identity``; otherwise the bearer middleware's current principal, + then a verified signed-request principal. A decisioning ``RequestContext`` + never contributes its ``caller_identity``, which is a cache key. + """ + from adcp.server.auth import current_principal + + if hasattr(context, "account_id"): + value = getattr(context, "caller_identity", None) + return value if isinstance(value, str) and value else None + principal = current_principal.get() + if principal: + return principal + auth_info = getattr(context, "auth_info", None) + if auth_info is None: + metadata = getattr(context, "metadata", None) + if isinstance(metadata, Mapping): + auth_info = metadata.get(_AUTH_INFO_KEY) + auth_principal = getattr(context, "auth_principal", None) + verified = getattr(auth_info, "principal", None) + if auth_principal is not None and auth_principal == verified: + return str(auth_principal) + if type(context).__name__ != "RequestContext": + # A transport ToolContext: the verified principal the decisioning + # RequestContext would expose as auth_principal, else the identity the + # context factory installed. + if isinstance(verified, str) and verified: + return verified + value = getattr(context, "caller_identity", None) + if isinstance(value, str) and value: + return value + return None + + +def _auth_snapshot(context: Any) -> dict[str, Any] | None: + """A token-free copy of the request's verified ``AuthInfo``, as JSON.""" + metadata = getattr(context, "metadata", None) + auth_info = metadata.get(_AUTH_INFO_KEY) if isinstance(metadata, Mapping) else None + if auth_info is None: + auth_info = getattr(context, "auth_info", None) + if isinstance(auth_info, Mapping): + # The legacy dict shape the decisioning handler also accepts. + from adcp.decisioning.context import AuthInfo + + auth_info = AuthInfo._from_legacy_dict(dict(auth_info)) + if auth_info is None or not hasattr(auth_info, "kind"): + return None + snapshot: dict[str, Any] = {"kind": str(auth_info.kind)} + for name in ("key_id", "principal", "agent_url", "operator"): + value = getattr(auth_info, name, None) + if isinstance(value, str) and value: + snapshot[name] = value + scopes = getattr(auth_info, "scopes", None) + if scopes: + snapshot["scopes"] = [str(item) for item in scopes] + credential: Any = getattr(auth_info, "credential", None) + kind = getattr(credential, "kind", None) + # Credentials here are identities, not secrets: a key id, a client id, a + # verified signer. Nothing else on AuthInfo (``extra``) is kept. + if kind == "api_key": + snapshot["credential"] = {"kind": kind, "key_id": credential.key_id} + elif kind == "oauth": + snapshot["credential"] = { + "kind": kind, + "client_id": credential.client_id, + "scopes": list(credential.scopes), + } + elif kind == "http_sig": + snapshot["credential"] = { + "kind": kind, + "keyid": credential.keyid, + "agent_url": credential.agent_url, + "verified_at": credential.verified_at, + } + return snapshot + + +def _restore_auth(snapshot: Mapping[str, Any] | None) -> Any: + if not snapshot: + return None + from adcp.decisioning.context import AuthInfo + from adcp.decisioning.registry import ApiKeyCredential, HttpSigCredential, OAuthCredential + + raw = dict(snapshot.get("credential") or {}) + credential: Any = None + if raw.get("kind") == "api_key": + credential = ApiKeyCredential(kind="api_key", key_id=raw["key_id"]) + elif raw.get("kind") == "oauth": + credential = OAuthCredential( + kind="oauth", client_id=raw["client_id"], scopes=tuple(raw.get("scopes") or ()) + ) + elif raw.get("kind") == "http_sig": + credential = HttpSigCredential( + kind="http_sig", + keyid=raw["keyid"], + agent_url=raw["agent_url"], + verified_at=float(raw["verified_at"]), + ) + return AuthInfo( + kind=str(snapshot["kind"]), + key_id=snapshot.get("key_id"), + principal=snapshot.get("principal"), + scopes=list(snapshot.get("scopes") or ()), + credential=credential, + agent_url=snapshot.get("agent_url"), + operator=snapshot.get("operator"), + ) + + +def _roster_unavailable(principal: PrincipalRef, cause: str) -> ReliableReportingConfigurationError: + return ReliableReportingConfigurationError( + f"the accounts principal {principal.principal_id!r} may read could not be listed " + f"under its identity ({cause}); reporting subscriptions never read that as an empty " + "roster. Supply SubscriptionOptions.accounts to list them another way.", + code="SUBSCRIPTION_ROSTER_UNAVAILABLE", + ) + + +# -- derived host -------------------------------------------------------------- + + +class _ManagedHost: + """The engine's host facts, derived from the installed platform or overridden.""" + + def __init__(self, managed: ManagedSubscriptions) -> None: + self._managed = managed + options = managed.options + if options.account_delivery_configs is not None: + self.account_delivery_configs = self._account_delivery_configs + if options.is_period_delivered is not None: + self.is_period_delivered = self._is_period_delivered + + async def list_authorized_accounts( + self, principal: Any, *, cursor: str | None + ) -> AuthorizedAccountPage: + ref = PrincipalRef.coerce(principal) + override = self._managed.options.accounts + if override is not None: + page = await _resolve(override(ref, cursor=cursor)) + if not isinstance(page, AuthorizedAccountPage): + raise TypeError("SubscriptionOptions.accounts must return AuthorizedAccountPage") + return page + return await self._managed._platform_accounts(ref, cursor) + + async def account_buys(self, principal: Any, account_id: str) -> Sequence[SubscriptionBuy]: + ref = PrincipalRef.coerce(principal) + override = self._managed.options.buys + if override is not None: + return tuple(await _resolve(override(ref, account_id))) + return await self._managed._platform_buys(ref, account_id) + + async def _account_delivery_configs( + self, principal: Any, account_id: str + ) -> Sequence[AccountDeliveryConfig]: + hook = self._managed.options.account_delivery_configs + assert hook is not None + return tuple(await _resolve(hook(PrincipalRef.coerce(principal), account_id))) + + async def _is_period_delivered( + self, + principal: Any, + account_id: str, + period: ExpectedSubscriptionPeriod, + *, + by: datetime, + ) -> bool: + hook = self._managed.options.is_period_delivered + assert hook is not None + return bool(await _resolve(hook(PrincipalRef.coerce(principal), account_id, period, by=by))) + + +@dataclass +class _Platform: + """Where the derived host reads ``list_accounts`` and ``get_media_buys``.""" + + target: Any = None + #: A decisioning ``PlatformHandler`` once ``serve`` builds it. + handler: Any = None + decisioning: bool = False + sync_accounts_seen: set[str] = field(default_factory=set) + + +class ManagedSubscriptions: + """The SDK-owned composition behind ``ReliableReportingService(subscriptions=...)``. + + Built by the service; sellers configure it through + :class:`SubscriptionOptions` and never construct it. + """ + + def __init__( + self, + options: SubscriptionOptions, + *, + clock: Callable[[], datetime], + status_retention_days: int, + report: Callable[[SubscriptionPlanningFailure], Awaitable[None]], + ) -> None: + if not isinstance(options, SubscriptionOptions): + raise TypeError("subscriptions must be SubscriptionOptions") + self.options = options + self._clock = clock + self._retention_days = status_retention_days + self._report = report + self._platform = _Platform() + self._engine: ReportingSubscriptionEngine | None = None + self._principals: PrincipalService | None = options.principals + self._service_options: ReportingSubscriptionServiceOptions | None = None + + # -- composition ----------------------------------------------------------- + + @property + def composed(self) -> bool: + return self._service_options is not None + + @property + def engine(self) -> ReportingSubscriptionEngine: + if self._engine is None: + raise ReliableReportingConfigurationError( + "reporting subscriptions are composed at install() or startup" + ) + return self._engine + + @property + def principals(self) -> PrincipalService: + if self._principals is None: + raise ReliableReportingConfigurationError( + "reporting subscriptions are composed at install() or startup" + ) + return self._principals + + @property + def service_options(self) -> ReportingSubscriptionServiceOptions: + assert self._service_options is not None + return self._service_options + + def compose( + self, offerings: Sequence[Mapping[str, Any]] + ) -> ReportingSubscriptionServiceOptions: + """Build the engine and the principal runtime from the registered offerings.""" + if self._service_options is not None: + return self._service_options + from adcp.reporting.service_subscriptions import ReportingSubscriptionServiceOptions + from adcp.reporting.subscriptions.capabilities import ReportingSubscriptionSupport + from adcp.reporting.subscriptions.engine import ReportingSubscriptionEngine + from adcp.server.principal import PrincipalService + + options = self.options + mapped = self._offerings(offerings) + dimensions = ( + tuple(options.dimensions) + if options.dimensions is not None + else tuple(sorted({item for offering in mapped for item in offering.dimensions})) + ) + try: + support = ReportingSubscriptionSupport( + max_accounts=options.max_accounts, + backfill_on_regrant=options.backfill_on_regrant, + dimensions=dimensions, + account_configuration=options.account_configuration, + ) + except ValueError as error: + raise ReliableReportingConfigurationError(str(error)) from None + if self._principals is None: + self._principals = PrincipalService( + options.principal_store, + emit_change=options.emit_change, + accepted_experimental_features=("protocol.principal",), + ) + try: + engine = ReportingSubscriptionEngine( + principals=self._principals, + host=_ManagedHost(self), + offerings=mapped, + support=support, + status_retention_days=self._retention_days, + store=options.state_store, + clock=self._clock, + end_account_configs=options.end_account_configs, + ) + except ValueError as error: + raise ReliableReportingConfigurationError(str(error)) from None + self._principals.set_reporting_subscriptions(engine.principal_options()) + self._engine = engine + self._service_options = ReportingSubscriptionServiceOptions( + engine=engine, + list_principals=self.list_principals, + resolve_principal=self.resolve, + planning_error_handler=options.planning_error_handler, + definitions=options.definitions, + bind_managed_generation=options.bind_managed_generation, + ) + return self._service_options + + @staticmethod + def _offerings(offerings: Sequence[Mapping[str, Any]]) -> list[SubscriptionOffering]: + from adcp.reporting.subscriptions._clock import CoverageInputError + from adcp.reporting.subscriptions.resolution import SubscriptionOffering + + if not offerings: + raise ReliableReportingConfigurationError( + "reporting subscriptions resolve against the service's registered offerings, and " + "none is registered. Register each source with capability_offerings, publishing " + "standard_dimensions on every offering.", + code="SUBSCRIPTIONS_REQUIRE_STANDARD_DIMENSIONS", + ) + mapped: list[SubscriptionOffering] = [] + for offering in offerings: + if "standard_dimensions" not in offering: + raise ReliableReportingConfigurationError( + f"offering {offering.get('offering_id')!r} has no standard_dimensions, so " + "a buyer cannot resolve a subscription selector against it. Publish " + "standard_dimensions (empty for totals only) in its capability_offerings " + "entry.", + code="SUBSCRIPTIONS_REQUIRE_STANDARD_DIMENSIONS", + ) + try: + mapped.append(SubscriptionOffering.from_offering(offering)) + except (CoverageInputError, KeyError, ValueError) as error: + raise ReliableReportingConfigurationError( + f"offering {offering.get('offering_id')!r} cannot serve subscriptions: {error}" + ) from None + return mapped + + def check_roster(self) -> None: + """Refuse a composition with no way to enumerate a principal's accounts.""" + if self.options.accounts is not None: + return + target = self._platform.target + if target is None: + return + if self._platform.decisioning: + listing = getattr(getattr(target, "accounts", None), "list", None) + available = callable(listing) + else: + from adcp.server import ADCPHandler + + method = getattr(type(target), "list_accounts", None) + available = method is not None and method is not ADCPHandler.list_accounts + if not available: + raise ReliableReportingConfigurationError( + "all_authorized_accounts subscriptions follow the accounts a principal may read, " + "and this platform does not serve list_accounts. Implement list_accounts (a " + "decisioning platform's accounts.list), or supply SubscriptionOptions.accounts.", + code="SUBSCRIPTION_ROSTER_UNAVAILABLE", + ) + if self.options.buys is None and not self._platform.decisioning: + from adcp.server import ADCPHandler + + method = getattr(type(target), "get_media_buys", None) + if method is None or method is ADCPHandler.get_media_buys: + raise ReliableReportingConfigurationError( + "reporting subscriptions read each account's buys through get_media_buys, " + "which this platform does not serve. Implement get_media_buys, or supply " + "SubscriptionOptions.buys." + ) + + def bind_platform(self, platform: Any, *, decisioning: bool) -> None: + self._platform.target = platform + self._platform.decisioning = decisioning + self.check_roster() + + def bind_handler(self, handler: Any) -> None: + """Record the decisioning ``PlatformHandler`` that serves the platform.""" + self._platform.handler = handler + + # -- principals ------------------------------------------------------------ + + async def list_principals(self) -> list[PrincipalRef]: + """Every principal with subscription state, from the engine store's index.""" + return await self.engine.list_principals() + + async def resolve(self, request: Any, context: Any) -> PrincipalRef: + """The authenticated principal of a request; never derived from request content.""" + tenant_id = getattr(context, "tenant_id", None) + override = self.options.resolve_principal + if override is not None: + value = await _resolve(override(request, context)) + return PrincipalRef.coerce(value, tenant_id=tenant_id) + subject = request_principal_id(context) + if not subject: + raise ReliableReportingConfigurationError( + "reporting subscriptions need an authenticated caller; supply " + "SubscriptionOptions.resolve_principal for another trust model", + kind="auth_required", + ) + return PrincipalRef( + tenant_id if isinstance(tenant_id, str) and tenant_id else None, subject + ) + + def _identity(self, principal: PrincipalRef) -> Any: + from adcp.server.principal import PrincipalIdentity + + return PrincipalIdentity(principal.principal_id, self.options.principal_kind) + + async def _remember(self, principal: PrincipalRef, context: Any) -> None: + binding: dict[str, Any] = {"tenant_id": principal.tenant_id} + snapshot = _auth_snapshot(context) + if snapshot is not None: + binding["auth"] = snapshot + await self.engine.record_binding(principal, binding) + + def _service_context(self, principal: PrincipalRef, binding: Mapping[str, Any] | None) -> Any: + """A request context for reads made on the principal's behalf outside a request.""" + from adcp.server.base import ToolContext + + metadata: dict[str, Any] = {"adcp.reporting.service_context": True} + auth = _restore_auth((binding or {}).get("auth")) + if auth is not None: + metadata[_AUTH_INFO_KEY] = auth + return ToolContext( + caller_identity=principal.principal_id, + tenant_id=principal.tenant_id, + metadata=metadata, + ) + + # -- served principal tasks ------------------------------------------------ + + async def get_principal(self, params: Any, context: Any) -> Any: + principal = await self.resolve(params, context) + return await self.principals.get_principal( + self._identity(principal), params, context=context + ) + + async def sync_principal(self, params: Any, context: Any) -> Any: + principal = await self.resolve(params, context) + await self._remember(principal, context) + response = await self.principals.sync_principal( + self._identity(principal), params, context=context + ) + result = getattr(response, "result", None) + if ( + self.options.reconcile_on_write + and getattr(result, "kind", None) == "applied" + and str(getattr(result, "action", "")) != "unchanged" + ): + await self.reconcile(principal) + return response + + # -- scheduling ------------------------------------------------------------ + + async def _subscribed(self, principal: PrincipalRef) -> bool: + return bool(await self.principals.reporting_subscriptions(principal)) + + async def reconcile(self, principal: PrincipalRef, *, now: datetime | None = None) -> bool: + """Reconcile one principal, isolating and reporting any failure.""" + from adcp.reporting.service_subscriptions import SubscriptionPlanningFailure + + try: + if not await self._subscribed(principal): + return True + await self.engine.reconcile(principal, now=now) + except Exception as error: # noqa: BLE001 - a trigger never fails the request + await self._report(SubscriptionPlanningFailure(principal, error)) + return False + return True + + async def touch(self, *, account_id: str | None, context: Any | None = None) -> None: + """Reconcile the principals a served write on ``account_id`` concerns. + + The caller, when it holds subscriptions, and every principal whose + engine state includes the account. + """ + if not self.options.reconcile_on_write or not self.composed: + return + concerned: dict[str, PrincipalRef] = {} + if context is not None: + try: + caller = await self.resolve(None, context) + except ReliableReportingConfigurationError: + caller = None + if caller is not None: + concerned[caller.subject] = caller + if account_id is not None: + try: + principals = await self.list_principals() + except Exception as error: # noqa: BLE001 - reported, never raised to the buyer + logger.warning("reporting subscription principals could not be listed: %s", error) + principals = [] + for principal in principals: + if account_id in await self.engine.account_ids(principal): + concerned.setdefault(principal.subject, principal) + for principal in concerned.values(): + await self.reconcile(principal) + + async def sweep(self, *, now: datetime) -> list[SubscriptionPlanningFailure]: + """The worker's boundary sweep: reconcile each principal a boundary passed for.""" + from adcp.reporting.service_subscriptions import SubscriptionPlanningFailure + + failures: list[SubscriptionPlanningFailure] = [] + for principal in await self.list_principals(): + try: + if not await self._subscribed(principal): + continue + if not await self.engine.boundary_due(principal, now=now): + continue + await self.engine.reconcile(principal, now=now) + except Exception as error: # noqa: BLE001 - isolate one principal + failures.append(SubscriptionPlanningFailure(principal, error)) + return failures + + # -- derived roster and buys ----------------------------------------------- + + async def _call(self, tool: str, request: Any, principal: PrincipalRef) -> dict[str, Any]: + binding = await self.engine.binding(principal) + context = self._service_context(principal, binding) + platform = self._platform + if platform.decisioning: + if platform.handler is None: + raise ReliableReportingConfigurationError( + "reporting subscriptions read the platform through the server that serves " + "it; pass the installed platform to adcp.decisioning.serve before reconciling" + ) + target = platform.handler + elif platform.target is not None: + target = platform.target + else: + raise ReliableReportingConfigurationError( + "reporting subscriptions derive accounts and buys from the installed platform; " + "call service.install(platform) first" + ) + response = _wire(await _resolve(getattr(target, tool)(request, context))) + if not isinstance(response, Mapping): + raise TypeError(f"{tool} returned {type(response).__name__}") + if response.get("supported") is False or response.get("errors"): + errors = response.get("errors") or [response.get("error") or {}] + codes = sorted({str((item or {}).get("code", "UNKNOWN")) for item in errors}) + raise RuntimeError(f"{tool} answered {', '.join(codes)}") + return dict(response) + + async def _platform_accounts( + self, principal: PrincipalRef, cursor: str | None + ) -> AuthorizedAccountPage: + from adcp.types import ListAccountsRequest + + request = ListAccountsRequest.model_validate( + {"pagination": {"cursor": cursor}} if cursor else {} + ) + try: + response = await self._call("list_accounts", request, principal) + except ReliableReportingConfigurationError: + raise + except Exception as error: # noqa: BLE001 - becomes a coded configuration error + raise _roster_unavailable(principal, type(error).__name__) from error + now = self._clock() + accounts: list[AuthorizedAccount] = [] + for raw in response.get("accounts") or (): + account = dict(raw) + account_id = account.get("account_id") + if not account_id: + continue + zone = account.get("timezone") + if not zone: + raise ReliableReportingConfigurationError( + f"account {account_id!r} has no IANA timezone, and coverage never " + "substitutes one. Serve timezone on every account list_accounts returns.", + code="SUBSCRIPTION_ROSTER_UNAVAILABLE", + ) + authorization = account.get("authorization") + tasks = None + authorized_at = now + if isinstance(authorization, Mapping): + if authorization.get("allowed_tasks") is not None: + tasks = frozenset(str(item) for item in authorization["allowed_tasks"]) + authorized_at = _instant(authorization.get("authorized_at")) or now + accounts.append( + AuthorizedAccount( + account_id=str(account_id), + timezone=str(zone), + authorized_at=authorized_at, + operator=account.get("operator"), + allowed_tasks=tasks, + ) + ) + pagination = response.get("pagination") or {} + next_cursor = pagination.get("cursor") if pagination.get("has_more") else None + return AuthorizedAccountPage(tuple(accounts), next_cursor=next_cursor) + + async def _platform_buys( + self, principal: PrincipalRef, account_id: str + ) -> tuple[SubscriptionBuy, ...]: + from adcp.types import GetMediaBuysRequest + + buys: list[SubscriptionBuy] = [] + cursor: str | None = None + for _page in range(_MAX_PAGES): + payload: dict[str, Any] = { + "account": {"account_id": account_id}, + "status_filter": list(_ALL_BUY_STATUSES), + } + if cursor: + payload["pagination"] = {"cursor": cursor} + response = await self._call( + "get_media_buys", GetMediaBuysRequest.model_validate(payload), principal + ) + for raw in response.get("media_buys") or (): + buy = self._buy(dict(raw)) + if buy is not None: + buys.append(buy) + pagination = response.get("pagination") or {} + cursor = pagination.get("cursor") if pagination.get("has_more") else None + if not cursor: + return tuple(buys) + raise RuntimeError(f"get_media_buys for account {account_id!r} did not finish paging") + + def _buy(self, raw: Mapping[str, Any]) -> SubscriptionBuy | None: + start, end = _instant(raw.get("start_time")), _instant(raw.get("end_time")) + if start is None or end is None: + return None # not yet flighted: nothing can be owed for it + cancellation = raw.get("cancellation") or {} + offerings = self.options.buy_offerings + if offerings is not None: + by_package = {str(key): tuple(value) for key, value in offerings(raw).items()} + else: + every = tuple(item.offering_id for item in self.engine.offerings) + by_package = { + str(package["package_id"]): every + for package in raw.get("packages") or () + if package.get("package_id") + } + packages = tuple( + SubscriptionBuyPackage(package_id, ids) + for package_id, ids in sorted(by_package.items()) + ) + return SubscriptionBuy( + media_buy_id=str(raw["media_buy_id"]), + confirmed_at=_instant(raw.get("confirmed_at")), + start_time=start, + end_time=end, + canceled_at=_instant(cancellation.get("canceled_at")), + completed_at=_instant(raw.get("completed_at")), + packages=packages, + reporting_delivery_offering_ids=( + () if packages else tuple(item.offering_id for item in self.engine.offerings) + ), + ) + + # -- capabilities ---------------------------------------------------------- + + @staticmethod + def advertise_principal(payload: dict[str, Any]) -> dict[str, Any]: + """Advertise the principal layer the service serves, merged with the seller's.""" + adcp = dict(payload.get("adcp") or {}) + principal = dict(adcp.get("principal") or {}) + principal.setdefault("supported", True) + principal.setdefault("sync_task", "sync_principal") + principal.setdefault("read_task", "get_principal") + sections = [str(item) for item in principal.get("supported_sections") or ()] + for section in ("reporting_destinations", "reporting_subscriptions"): + if section not in sections: + sections.append(section) + principal["supported_sections"] = sections + adcp["principal"] = principal + payload = {**payload, "adcp": adcp} + experimental = [str(item) for item in payload.get("experimental_features") or ()] + if "protocol.principal" not in experimental: + experimental.append("protocol.principal") + payload["experimental_features"] = experimental + return payload + + +__all__ = [ + "AccountConfigsOverride", + "AccountsOverride", + "BuyOfferingsOverride", + "BuysOverride", + "ManagedPrincipalResolver", + "ManagedSubscriptions", + "PeriodDeliveredOverride", + "SubscriptionOptions", + "request_principal_id", +] diff --git a/src/adcp/reporting/subscriptions/registry.py b/src/adcp/reporting/subscriptions/registry.py index 9eb800951..38a48f0f1 100644 --- a/src/adcp/reporting/subscriptions/registry.py +++ b/src/adcp/reporting/subscriptions/registry.py @@ -107,6 +107,7 @@ def _validate_one( *, destinations: Mapping[str, Mapping[str, Any]], delivery_config_ids: Collection[str], + replace_account_configs: bool = False, ) -> None: where = f"reporting_subscriptions[{index}]" subscription_id = configuration["subscription_id"] @@ -130,9 +131,10 @@ def _validate_one( raise ReportingSubscriptionValidationError( f"{where}.exclude_account_ids is valid only with all_authorized_accounts scope" ) - if configuration.get("replace_account_configs") is True: - # Not applied yet: accepting it would leave the caller's account-level - # configurations in force while it believes they were cut over. + if configuration.get("replace_account_configs") is True and not replace_account_configs: + # Accepting it without a runtime that applies it would leave the + # caller's account-level configurations in force while it believes + # they were cut over. raise ReportingSubscriptionValidationError( f"{where}.replace_account_configs is not supported by this seller", code="UNSUPPORTED_FEATURE", @@ -254,12 +256,15 @@ def apply_reporting_subscriptions( destinations: Sequence[Any], now: datetime, delivery_config_ids: Collection[str] = (), + replace_account_configs: bool = False, ) -> list[ReportingSubscriptionState]: """Return the new subscription states for a ``sync_principal`` request. ``destinations`` are the caller's ``reporting_destinations`` states after this request applies. ``delivery_config_ids`` are every ``delivery_config_id`` the caller uses through ``sync_accounts``. + ``replace_account_configs`` accepts that flag; only a runtime that applies + it passes ``True``, and otherwise it is refused with ``UNSUPPORTED_FEATURE``. Engine-derived fields (``accounts_with_exceptions``, ``accounts_unauthorized``, ``issues``) carry over from ``previous``. """ @@ -281,6 +286,7 @@ def apply_reporting_subscriptions( index, destinations=targets, delivery_config_ids=delivery_config_ids, + replace_account_configs=replace_account_configs, ) _validate_set([item for item in configurations if item.get("active", True)], targets) diff --git a/src/adcp/server/__init__.py b/src/adcp/server/__init__.py index a8fe597f5..c260732e8 100644 --- a/src/adcp/server/__init__.py +++ b/src/adcp/server/__init__.py @@ -52,6 +52,7 @@ async def get_products(params, context=None): from __future__ import annotations +from adcp._principal_ref import PrincipalRef from adcp.capabilities import validate_capabilities from adcp.server._hooks import ( PreValidationHook, @@ -252,6 +253,7 @@ async def get_products(params, context=None): "PrincipalChange", "PrincipalChangeEmitter", "PrincipalIdentity", + "PrincipalRef", "PrincipalRecord", "PrincipalRecordStore", "PrincipalService", diff --git a/src/adcp/server/principal.py b/src/adcp/server/principal.py index bd162b7cc..fb6a70feb 100644 --- a/src/adcp/server/principal.py +++ b/src/adcp/server/principal.py @@ -22,6 +22,7 @@ from pydantic import BaseModel, TypeAdapter from adcp._deferred_adapters import deferred_adapter +from adcp._principal_ref import PrincipalRef from adcp._version import normalize_to_release_precision from adcp.server.idempotency import canonical_json_sha256 from adcp.types import ( @@ -126,6 +127,10 @@ class ReportingSubscriptionOptions: delivery_config_ids: DeliveryConfigIdsHook | None = None record_versions: SubscriptionVersionsHook | None = None clock: Callable[[], datetime] | None = None + #: Accept ``replace_account_configs: true``. Only a runtime that applies it + #: (an engine built with ``end_account_configs``) sets this; otherwise the + #: flag is refused with ``UNSUPPORTED_FEATURE``. + replace_account_configs: bool = False @dataclass @@ -1015,6 +1020,7 @@ async def _replace_subscriptions( destinations=destinations, now=self._now(), delivery_config_ids=await hook(identity) if hook else (), + replace_account_configs=self._subscriptions.replace_account_configs, ) elif prior and "reporting_destinations" in desired.model_fields_set: names = {_destination_id(item) for item in destinations} @@ -1040,7 +1046,7 @@ def _now(self) -> datetime: async def update_subscription_status( self, - identity: PrincipalIdentity, + identity: PrincipalIdentity | PrincipalRef, subscription_id: str, *, accounts_with_exceptions: Sequence[Mapping[str, Any]] | None = None, @@ -1104,7 +1110,7 @@ async def update_subscription_status( return True async def reporting_subscriptions( - self, identity: PrincipalIdentity + self, identity: PrincipalIdentity | PrincipalRef ) -> list[ReportingSubscriptionState]: """The caller's stored subscription states, for the reporting runtime.""" record = await self._store.get(identity.subject) @@ -1115,7 +1121,9 @@ async def reporting_subscriptions( for item in record.configuration.reporting_subscriptions or [] ] - async def reporting_subscription_ids(self, identity: PrincipalIdentity) -> set[str]: + async def reporting_subscription_ids( + self, identity: PrincipalIdentity | PrincipalRef + ) -> set[str]: """Every subscription_id the caller holds, inactive ones included. ``sync_accounts`` must reject a ``delivery_config_id`` equal to one of @@ -1124,7 +1132,7 @@ async def reporting_subscription_ids(self, identity: PrincipalIdentity) -> set[s return {item.subscription_id for item in await self.reporting_subscriptions(identity)} async def reporting_destinations( - self, identity: PrincipalIdentity + self, identity: PrincipalIdentity | PrincipalRef ) -> list[AgentReportingDestinationState]: """The caller's stored destination states, for the reporting runtime.""" record = await self._store.get(identity.subject) diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 61642ee41..09d8d80a0 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -765,6 +765,7 @@ "FaultInjectingReportingStagingStore": "adcp.reporting._testing_failures", "InMemoryConsumerStatusCheckpoints": "adcp.reporting._consumer", "ObligationReconciliation": "adcp.reporting._reconcile", + "PrincipalRef": "adcp._principal_ref", "ReliableReportingConfigurationError": "adcp.reporting._errors", "ReliableReportingService": "adcp.reporting.service", "ReliableReportingServiceError": "adcp.reporting.service_lifecycle", @@ -791,9 +792,11 @@ "ReportingServiceOffering": "adcp.reporting.service_production", "ReportingServiceResource": "adcp.reporting.service_lifecycle", "ReportingStatusClient": "adcp.reporting._reconcile", + "ReportingSubscriptionServiceOptions": "adcp.reporting.service_subscriptions", "ReportingTestBarrier": "adcp.reporting._testing_failures", "ReportingTier": "adcp.reporting._reconcile", "ScriptedReportingAdapter": "adcp.reporting.testing", + "SubscriptionOptions": "adcp.reporting.subscriptions.managed", "assert_reporting_lifecycle_replay": "adcp.reporting._testing_lifecycle", "build_reporting_receipt": "adcp.reporting._reconcile", "capture_reporting_lifecycle": "adcp.reporting._testing_lifecycle", diff --git a/tests/test_decisioning_handler_shims.py b/tests/test_decisioning_handler_shims.py index 94d97f60e..ce4edbdf5 100644 --- a/tests/test_decisioning_handler_shims.py +++ b/tests/test_decisioning_handler_shims.py @@ -80,6 +80,9 @@ def test_advertised_tools_covers_every_specialism_wire_tool() -> None: "get_reporting_status", "sync_reporting_status", "sync_reporting_receipts", + # Managed reporting subscriptions (advertised only when the service serves them) + "get_principal", + "sync_principal", # AdCP 3.2 compact lifecycle (advertised subset is capability-filtered) "list_products", "request_proposals", diff --git a/tests/test_reporting_subscription_delivery_authority.py b/tests/test_reporting_subscription_delivery_authority.py index ace1b6bd1..a7e53eabb 100644 --- a/tests/test_reporting_subscription_delivery_authority.py +++ b/tests/test_reporting_subscription_delivery_authority.py @@ -11,6 +11,7 @@ import pytest +from adcp.reporting import PrincipalRef from adcp.reporting.ledger import ReportingConfigurationGenerationKey from adcp.reporting.materializer.contracts import ReportingWriterError from adcp.reporting.materializer.service import ReportingMaterializerService @@ -44,10 +45,10 @@ def _target(offering_id: str | None): @pytest.mark.asyncio async def test_account_generations_never_consult_the_subscription_authority() -> None: - calls: list[tuple[str, str]] = [] + calls: list[tuple[PrincipalRef, str]] = [] - async def authorize(subject: str, account_id: str) -> bool: - calls.append((subject, account_id)) + async def authorize(principal: PrincipalRef, account_id: str) -> bool: + calls.append((principal, account_id)) return False await _service(authorize)._require_subscription_authority( @@ -59,8 +60,8 @@ async def authorize(subject: str, account_id: str) -> bool: @pytest.mark.asyncio @pytest.mark.parametrize("effect", ["not_started", "unknown"]) async def test_a_lost_grant_holds_the_attempt_under_the_same_identity(effect: str) -> None: - async def authorize(subject: str, account_id: str) -> bool: - assert (subject, account_id) == (AGENCY.subject, ACCOUNT) + async def authorize(principal: PrincipalRef, account_id: str) -> bool: + assert (principal, account_id) == (PrincipalRef(None, AGENCY.subject), ACCOUNT) return False with pytest.raises(ReportingWriterError) as held: @@ -78,7 +79,7 @@ async def authorize(subject: str, account_id: str) -> bool: @pytest.mark.asyncio async def test_a_current_grant_delivers() -> None: - async def authorize(subject: str, account_id: str) -> bool: + async def authorize(principal: PrincipalRef, account_id: str) -> bool: return True await _service(authorize)._require_subscription_authority( @@ -90,7 +91,7 @@ async def authorize(subject: str, account_id: str) -> bool: async def test_a_hung_hook_is_bounded_by_the_attempt_deadline() -> None: import asyncio - async def authorize(subject: str, account_id: str) -> bool: + async def authorize(principal: PrincipalRef, account_id: str) -> bool: await asyncio.sleep(30) return True @@ -112,7 +113,8 @@ async def test_the_engine_answers_from_the_grant() -> None: [_buy("media-buy-a1", "2026-11-01T01:00:00Z", "2026-11-01T03:00:00Z")], allowed_tasks=frozenset({"get_reporting_status"}), ) - assert await seller.engine.authorizes_delivery(AGENCY.subject, ACCOUNT) is True + principal = PrincipalRef(None, AGENCY.subject) + assert await seller.engine.authorizes_delivery(principal, ACCOUNT) is True del seller.host.accounts[ACCOUNT] await seller.engine.reconcile(AGENCY) - assert await seller.engine.authorizes_delivery(AGENCY.subject, ACCOUNT) is False + assert await seller.engine.authorizes_delivery(principal, ACCOUNT) is False diff --git a/tests/test_reporting_subscription_managed.py b/tests/test_reporting_subscription_managed.py new file mode 100644 index 000000000..29af0cad6 --- /dev/null +++ b/tests/test_reporting_subscription_managed.py @@ -0,0 +1,529 @@ +"""The managed seller path: ``ReliableReportingService(subscriptions=SubscriptionOptions(...))``. + +The block's presence enables reporting subscriptions. The service owns the +principal runtime, serves ``get_principal``/``sync_principal`` from the +installed platform, derives the roster from ``list_accounts`` and the buys from +``get_media_buys`` under the principal's identity, reconciles on served writes, +and sweeps period boundaries in each worker turn. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any + +import pytest + +from adcp.reporting import PrincipalRef, SubscriptionOptions +from adcp.reporting.fixtures import SNAPSHOT_OFFERING_ID, redacted_capabilities +from adcp.reporting.ledger import ReportingConfiguration +from adcp.reporting.service import ( + ReliableReportingConfigurationError, + ReliableReportingService, + ReportingAccountContext, +) +from adcp.reporting.subscriptions import AccountDeliveryConfig +from adcp.reporting.testing import DeterministicReportingClock +from adcp.server import ADCPHandler +from adcp.server.base import AccountAwareToolContext +from adcp.server.mcp_tools import get_tools_for_handler +from tests.test_reliable_reporting_service import DEFINITION, _capability_offering +from tests.test_reporting_subscription_service import _ByBuySource + +START = datetime(2026, 11, 1, 0, 0, tzinfo=timezone.utc) +ACCOUNT = "account-redacted" +AGENCY = "https://agency.example/agent" +OTHER = "https://other.example/agent" +OFFERING = {**_capability_offering("gam"), "standard_dimensions": ["package"]} +SUBSCRIPTION = { + "subscription_id": "analytics-hourly", + "subscription_version": 1, + "active": True, + "scope": {"all_authorized_accounts": True}, + "selector": { + "feed_purpose": "analytics", + "period_duration": "PT1H", + "required_finality": "snapshot", + "dimensions": ["package"], + }, + "reconciliation_mode": "delivery_only", +} + + +def _context(configuration: ReportingConfiguration) -> ReportingAccountContext: + return ReportingAccountContext( + account_id=configuration.account_id, + adapter="gam", + currency="USD", + account_timezone="UTC", + source_scope=redacted_capabilities().source_scope, + snapshot_offering_id=SNAPSHOT_OFFERING_ID, + requested_dimensions=("campaign_id",), + capability_offering=OFFERING, + publication_namespace="reporting-source:fixture", + ) + + +def _ctx(principal: str = AGENCY) -> AccountAwareToolContext: + return AccountAwareToolContext(caller_identity=principal, account_id=ACCOUNT) + + +class Seller(ADCPHandler): + """A seller that already serves list_accounts and get_media_buys.""" + + def __init__(self) -> None: + self.accounts: dict[str, dict[str, Any]] = { + ACCOUNT: { + "account_id": ACCOUNT, + "name": "Redacted", + "status": "active", + "timezone": "UTC", + "operator": "agency.example", + "authorization": { + "allowed_tasks": ["get_reporting_status", "get_media_buys"], + "authorized_at": "2026-10-01T00:00:00Z", + }, + } + } + self.buys: list[dict[str, Any]] = [ + { + "media_buy_id": "media-buy-a1", + "status": "active", + "confirmed_at": "2026-11-01T01:00:00Z", + "start_time": "2026-11-01T01:00:00Z", + "end_time": "2026-11-01T03:00:00Z", + "packages": [{"package_id": "pkg-1", "product_id": "prod-1"}], + } + ] + self.listings: list[Any] = [] + self.buy_reads: list[Any] = [] + self.fail_listing = False + + async def get_adcp_capabilities(self, params: Any, context: Any = None) -> Any: + return {"adcp": {"major_versions": [3]}, "supported_protocols": ["media_buy"]} + + async def list_accounts(self, params: Any, context: Any = None) -> Any: + self.listings.append(context) + if self.fail_listing: + raise RuntimeError("the upstream token is only available inside a request") + return {"accounts": list(self.accounts.values())} + + async def get_media_buys(self, params: Any, context: Any = None) -> Any: + self.buy_reads.append((params, context)) + return {"media_buys": list(self.buys)} + + async def create_media_buy(self, params: Any, context: Any = None) -> Any: + return {"media_buy_id": "media-buy-new", "status": "active", "packages": []} + + +class _World: + def __init__(self, **options: Any) -> None: + self.clock = DeterministicReportingClock(START) + self.failures: list[Any] = [] + self.source = _ByBuySource() + options.setdefault("definitions", {OFFERING["offering_id"]: DEFINITION}) + options.setdefault("planning_error_handler", self.failures.append) + self.service = ReliableReportingService.memory( + account_context=_context, + clock=self.clock, + read_jitter_window=timedelta(0), + subscriptions=SubscriptionOptions(max_accounts=10, **options), + ) + self.service.sources.register("gam", self.source, capability_offerings=[OFFERING]) + self.seller = Seller() + self.platform = self.service.install(self.seller) + self._key = 0 + + async def subscribe(self, principal: str = AGENCY, **changes: Any) -> Any: + self._key += 1 + response = await self.platform.sync_principal( + { + "idempotency_key": f"managed-test-key-{self._key:06d}", + "configuration": {"reporting_subscriptions": [{**SUBSCRIPTION, **changes}]}, + }, + _ctx(principal), + ) + assert response.result.kind == "applied", response + return response + + +@pytest.mark.asyncio +async def test_the_block_alone_serves_subscriptions_end_to_end() -> None: + world = _World() + tools = {item["name"] for item in get_tools_for_handler(world.platform, _include_schemas=False)} + assert {"get_principal", "sync_principal", "get_reporting_status"} <= tools + + await world.subscribe() + # sync_principal reconciled at once: the roster came from list_accounts, + # read under the principal's identity. + assert [context.caller_identity for context in world.seller.listings] == [AGENCY] + principal = PrincipalRef(None, AGENCY) + assert await world.service._managed.engine.account_ids(principal) == {ACCOUNT} + + read = await world.platform.get_principal({}, _ctx()) + assert read.result.configuration.reporting_subscriptions[0].subscription_id == ( + "analytics-hourly" + ) + + world.clock.set(START + timedelta(hours=3, minutes=30)) + turn = await world.service.run_worker() + assert turn.subscription_errors == {} + # The boundary sweep reconciled before planning; both ended hours are owed. + assert turn.subscription_obligations_written == 2 + assert len(world.seller.listings) == 2 + status = await world.platform.get_reporting_status({"view": "periods"}, _ctx()) + assert status["subscription_coverage"][0]["subscription_id"] == "analytics-hourly" + assert {tuple(p["media_buy_ids"]) for p in status["periods"]} == {("media-buy-a1",)} + + +@pytest.mark.asyncio +async def test_capabilities_advertise_the_principal_layer_the_service_serves() -> None: + world = _World() + capabilities = await world.platform.get_adcp_capabilities({}, _ctx()) + principal = capabilities["adcp"]["principal"] + assert principal["sync_task"] == "sync_principal" and principal["read_task"] == "get_principal" + assert {"reporting_destinations", "reporting_subscriptions"} <= set( + principal["supported_sections"] + ) + assert {"protocol.principal", "media_buy.reporting_subscriptions"} <= set( + capabilities["experimental_features"] + ) + block = capabilities["media_buy"]["reporting_delivery"] + assert block["subscriptions"] == { + "supported": True, + "max_accounts": 10, + "dimensions": ["package"], + } + assert block["configuration_tasks"] == ["sync_principal", "sync_accounts"] + + +def test_every_offering_must_publish_standard_dimensions() -> None: + service = ReliableReportingService.memory( + account_context=_context, subscriptions=SubscriptionOptions() + ) + service.sources.register( + "gam", _ByBuySource(), capability_offerings=[_capability_offering("gam")] + ) + with pytest.raises(ReliableReportingConfigurationError) as refused: + service.install(Seller()) + error = refused.value + assert error.code == "SUBSCRIPTIONS_REQUIRE_STANDARD_DIMENSIONS" + assert error.recovery == "correctable" and error.dispatched is None + assert "capability_offerings" in str(error) + + +def test_a_platform_without_a_roster_is_refused_at_install() -> None: + class NoRoster(ADCPHandler): + async def get_media_buys(self, params: Any, context: Any = None) -> Any: + return {"media_buys": []} + + service = ReliableReportingService.memory( + account_context=_context, subscriptions=SubscriptionOptions() + ) + service.sources.register("gam", _ByBuySource(), capability_offerings=[OFFERING]) + with pytest.raises(ReliableReportingConfigurationError) as refused: + service.install(NoRoster()) + assert refused.value.code == "SUBSCRIPTION_ROSTER_UNAVAILABLE" + assert "SubscriptionOptions.accounts" in str(refused.value) + + +@pytest.mark.asyncio +async def test_a_failed_listing_is_reported_and_never_revokes() -> None: + world = _World() + await world.subscribe() + principal = PrincipalRef(None, AGENCY) + engine = world.service._managed.engine + assert await engine.grants_task(principal, ACCOUNT, "get_reporting_status") + + world.seller.fail_listing = True + world.clock.set(START + timedelta(hours=1, minutes=5)) + turn = await world.service.run_worker() + [error] = turn.subscription_errors.values() + assert isinstance(error, ReliableReportingConfigurationError) + assert error.code == "SUBSCRIPTION_ROSTER_UNAVAILABLE" + assert "SubscriptionOptions.accounts" in str(error) + assert [failure.principal for failure in world.failures] == [principal] + # The account was not read as revoked. + assert await engine.grants_task(principal, ACCOUNT, "get_reporting_status") + + +@pytest.mark.asyncio +async def test_the_sweep_reconciles_only_after_a_boundary() -> None: + world = _World() + await world.subscribe() + assert len(world.seller.listings) == 1 + world.clock.set(START + timedelta(minutes=30)) + await world.service.run_worker() + assert len(world.seller.listings) == 1 # no boundary since the last reconcile + world.clock.set(START + timedelta(hours=1)) + await world.service.run_worker() + assert len(world.seller.listings) == 2 + await world.service.run_worker() + assert len(world.seller.listings) == 2 + + +@pytest.mark.asyncio +async def test_served_writes_reconcile_the_principals_concerned() -> None: + world = _World() + await world.subscribe() + await world.subscribe(OTHER) + calls = len(world.seller.listings) + await world.platform.create_media_buy({"account": {"account_id": ACCOUNT}}, _ctx()) + # Both principals hold state for the account. + assert len(world.seller.listings) == calls + 2 + await world.service.touch(account_id="unknown-account") + assert len(world.seller.listings) == calls + 2 + + +@pytest.mark.asyncio +async def test_unchanged_sync_principal_does_not_reconcile() -> None: + world = _World() + await world.subscribe() + response = await world.platform.sync_principal( + { + "idempotency_key": "managed-test-key-unchanged", + "configuration": {"reporting_subscriptions": [SUBSCRIPTION]}, + }, + _ctx(), + ) + assert response.result.action == "unchanged" + assert len(world.seller.listings) == 1 + + +@pytest.mark.asyncio +async def test_overrides_replace_each_derivation() -> None: + from adcp.reporting.subscriptions import AuthorizedAccount, AuthorizedAccountPage + + seen: list[PrincipalRef] = [] + + async def accounts(principal: PrincipalRef, *, cursor: str | None) -> AuthorizedAccountPage: + seen.append(principal) + return AuthorizedAccountPage( + (AuthorizedAccount(ACCOUNT, timezone="UTC", authorized_at=START),) + ) + + world = _World(accounts=accounts, buys=lambda principal, account_id: []) + world.seller.fail_listing = True + await world.subscribe() + assert seen == [PrincipalRef(None, AGENCY)] + assert world.seller.listings == [] and world.seller.buy_reads == [] + + +@pytest.mark.asyncio +async def test_derived_buys_use_buy_offerings_per_package() -> None: + world = _World(buy_offerings=lambda buy: {"pkg-1": ["SOME_OTHER_OFFERING"]}) + await world.subscribe() + world.clock.set(START + timedelta(hours=3, minutes=30)) + turn = await world.service.run_worker() + # The package resolves no registered offering, so nothing is owed. + assert turn.subscription_obligations_written == 0 + + +@pytest.mark.asyncio +async def test_replace_account_configs_is_refused_without_a_host_that_applies_it() -> None: + world = _World() + response = await world.platform.sync_principal( + { + "idempotency_key": "managed-test-key-replace", + "configuration": { + "reporting_subscriptions": [{**SUBSCRIPTION, "replace_account_configs": True}] + }, + }, + _ctx(), + ) + assert response.result.kind == "failed" + assert [error.code for error in response.result.errors] == ["UNSUPPORTED_FEATURE"] + + +@pytest.mark.asyncio +async def test_replace_account_configs_applies_once_per_version_and_on_a_change_to_true() -> None: + configs = {ACCOUNT: [AccountDeliveryConfig("analytics-daily", "analytics")]} + ended: list[tuple[PrincipalRef, str, tuple[str, ...], datetime]] = [] + + async def end(principal, account_id, ids, *, at): + ended.append((principal, account_id, ids, at)) + + world = _World( + account_delivery_configs=lambda principal, account_id: configs.get(account_id, []), + end_account_configs=end, + ) + await world.subscribe(replace_account_configs=True) + assert ended == [(PrincipalRef(None, AGENCY), ACCOUNT, ("analytics-daily",), START)] + # Re-sending the same version and flag does nothing more. + world.clock.set(START + timedelta(hours=1)) + await world.service.run_worker() + assert len(ended) == 1 + # Clearing the flag, then setting it again, applies it again. + await world.subscribe(replace_account_configs=False) + await world.subscribe(replace_account_configs=True) + assert len(ended) == 2 + # A new version applies it again. + await world.subscribe( + subscription_version=2, + replace_account_configs=True, + operator_domains=["agency.example"], + ) + assert len(ended) == 3 + + +def test_end_account_configs_needs_the_configurations_it_ends() -> None: + async def end(principal, account_id, ids, *, at): + return None + + with pytest.raises(ValueError, match="account_delivery_configs"): + SubscriptionOptions(end_account_configs=end) + + +def test_managed_and_advanced_compositions_are_exclusive() -> None: + from adcp.reporting.service_subscriptions import ReportingSubscriptionServiceOptions + + with pytest.raises(ReliableReportingConfigurationError, match="not both"): + ReliableReportingService.memory( + account_context=_context, + subscriptions=SubscriptionOptions(), + reporting_subscriptions=object.__new__(ReportingSubscriptionServiceOptions), + ) + + +@pytest.mark.asyncio +async def test_the_binding_keeps_no_credential_material() -> None: + from adcp.decisioning import AuthInfo + from adcp.decisioning.registry import ApiKeyCredential + + world = _World() + context = _ctx() + context.metadata["adcp.auth_info"] = AuthInfo( + kind="api_key", + key_id="kid-1", + principal=AGENCY, + credential=ApiKeyCredential(kind="api_key", key_id="kid-1"), + extra={"session_secret": "do-not-store"}, + ) + await world.platform.sync_principal( + { + "idempotency_key": "managed-test-key-binding", + "configuration": {"reporting_subscriptions": [SUBSCRIPTION]}, + }, + context, + ) + binding = await world.service._managed.engine.binding(PrincipalRef(None, AGENCY)) + assert binding == { + "tenant_id": None, + "auth": { + "kind": "api_key", + "key_id": "kid-1", + "principal": AGENCY, + "credential": {"kind": "api_key", "key_id": "kid-1"}, + }, + } + # The roster is read under the restored identity. + restored = world.seller.listings[-1].metadata["adcp.auth_info"] + assert (restored.principal, restored.credential.key_id) == (AGENCY, "kid-1") + assert restored.extra == {} + + +# -- decisioning platform ------------------------------------------------------ + + +class _DecisioningAccounts: + resolution = "explicit" + + def __init__(self) -> None: + self.listed: list[Any] = [] + + def resolve(self, ref: Any, auth_info: Any = None) -> Any: + from adcp.decisioning import Account + + return Account(id=ACCOUNT, timezone="UTC") + + def list(self, filter: Any, ctx: Any = None) -> Any: + from adcp.decisioning import Account + + self.listed.append(ctx) + return [ + Account( + id=ACCOUNT, + timezone="UTC", + operator="agency.example", + authorization={ + "allowed_tasks": ["get_reporting_status", "get_media_buys"], + "authorized_at": "2026-10-01T00:00:00Z", + }, + ) + ] + + +def _decisioning_platform() -> Any: + from tests.test_lazy_platform_router import _SyncSalesPlatform + + class Platform(_SyncSalesPlatform): + def __init__(self) -> None: + super().__init__("managed") + self.accounts = _DecisioningAccounts() + + def get_media_buys(self, req: Any, ctx: Any) -> dict[str, Any]: + self.calls.append(("get_media_buys", ctx.account.id)) + return {"media_buys": list(Seller().buys)} + + return Platform() + + +@pytest.mark.asyncio +async def test_a_decisioning_platform_is_served_through_its_own_handler() -> None: + from adcp.decisioning.serve import create_adcp_server_from_platform + from adcp.server.auth import current_principal + from adcp.server.base import ToolContext + + clock = DeterministicReportingClock(START) + service = ReliableReportingService.memory( + account_context=_context, + clock=clock, + read_jitter_window=timedelta(0), + subscriptions=SubscriptionOptions(definitions={OFFERING["offering_id"]: DEFINITION}), + ) + service.sources.register("gam", _ByBuySource(), capability_offerings=[OFFERING]) + platform = _decisioning_platform() + installed = service.install(platform) + handler, _executor, _ = create_adcp_server_from_platform(installed, validate_at_init=False) + tools = {item["name"] for item in get_tools_for_handler(handler, _include_schemas=False)} + assert {"get_principal", "sync_principal"} <= tools + + token = current_principal.set(AGENCY) + try: + context = ToolContext(metadata={"adcp.auth_info": {"kind": "bearer", "principal": AGENCY}}) + response = await handler.sync_principal( + { + "idempotency_key": "managed-decisioning-key-1", + "configuration": {"reporting_subscriptions": [SUBSCRIPTION]}, + }, + context, + ) + assert response.result.kind == "applied" + # The roster was read through accounts.list under the restored identity. + [listing] = platform.accounts.listed + assert listing.auth_info.principal == AGENCY + assert ("get_media_buys", ACCOUNT) in platform.calls + + clock.set(START + timedelta(hours=3, minutes=30)) + turn = await service.run_worker() + assert turn.subscription_errors == {} + assert turn.subscription_obligations_written == 2 + + from adcp.types import CreateMediaBuyRequest + + await handler.create_media_buy( + CreateMediaBuyRequest( + account={"account_id": ACCOUNT}, + brand={"domain": "brand.example"}, + idempotency_key="managed-decisioning-buy-1", + start_time="2026-11-02T00:00:00Z", + end_time="2026-12-01T00:00:00Z", + packages=[ + {"buyer_ref": "pkg_1", "product_id": "prod_1", "pricing_option_id": "cpm"} + ], + ), + ToolContext(metadata={"adcp.auth_info": {"kind": "bearer", "principal": AGENCY}}), + ) + assert len(platform.accounts.listed) == 3 + finally: + current_principal.reset(token) diff --git a/tests/test_reporting_subscription_parity.py b/tests/test_reporting_subscription_parity.py index b05b2f774..cfe0c4938 100644 --- a/tests/test_reporting_subscription_parity.py +++ b/tests/test_reporting_subscription_parity.py @@ -326,3 +326,23 @@ def test_reason_codes_match_the_frozen_spec_vocabulary() -> None: assert dict(COVERAGE_REASON_CODES) == { item["code"]: item.get("rule") for item in vocabulary["reasons"] } + + +def test_seller_configuration_codes_match_the_frozen_spec_vocabulary() -> None: + from adcp.reporting._errors import ( + SELLER_CONFIGURATION_CODES, + ReliableReportingConfigurationError, + ) + + vocabulary = json.loads((SPEC / "error-codes.json").read_text()) + seller = { + entry["code"]: entry["recovery"] + for entry in vocabulary["codes"] + if entry["sdk_classes"]["python"] == "ReliableReportingConfigurationError" + } + assert set(seller) == SELLER_CONFIGURATION_CODES + for code, recovery in seller.items(): + error = ReliableReportingConfigurationError("x", code=code) + assert (error.code, error.recovery, error.dispatched) == (code, recovery, None) + with pytest.raises(ValueError): + ReliableReportingConfigurationError("x", code="NOT_A_CODE") From 7fc664af01fbca75995c087397b1230aa64b9e4d Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 19:34:46 +0000 Subject: [PATCH 2/5] fix(decisioning): type the managed principal task parameters The wire dispatcher resolves a shim's request model from its params annotation; get_principal and sync_principal on PlatformHandler now name GetPrincipalRequest and SyncPrincipalRequest, so dispatch validates them like every other task. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/decisioning/handler.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/adcp/decisioning/handler.py b/src/adcp/decisioning/handler.py index 0b1af0f01..f5e505cd0 100644 --- a/src/adcp/decisioning/handler.py +++ b/src/adcp/decisioning/handler.py @@ -186,6 +186,7 @@ class _PublicAccount(_DecisioningAccount[Any]): GetMediaBuysResponse, GetPlanAuditLogsRequest, GetPlanAuditLogsResponse, + GetPrincipalRequest, GetProductsRequest, GetProductsResponse, GetPropertyListRequest, @@ -227,6 +228,7 @@ class _PublicAccount(_DecisioningAccount[Any]): SyncCreativesSuccessResponse, SyncPlansRequest, SyncPlansResponse, + SyncPrincipalRequest, SyncReportingReceiptsRequest, SyncReportingStatusRequest, UpdateCollectionListRequest, @@ -3093,14 +3095,14 @@ async def _invoke_principal_method( name, lambda: getattr(reporting, name)(params, tool_ctx), decisioning=True ) - async def get_principal( - self, params: Any, context: ToolContext | None = None + async def get_principal( # type: ignore[override] + self, params: GetPrincipalRequest, context: ToolContext | None = None ) -> Any: """Served by an installed ``ReliableReportingService(subscriptions=...)``.""" return await self._invoke_principal_method("get_principal", params, context) - async def sync_principal( - self, params: Any, context: ToolContext | None = None + async def sync_principal( # type: ignore[override] + self, params: SyncPrincipalRequest, context: ToolContext | None = None ) -> Any: """Served by an installed ``ReliableReportingService(subscriptions=...)``.""" return await self._invoke_principal_method("sync_principal", params, context) From a4c9b6bad811405eb9f4e2e7f017303dcd6d781e Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 20:03:50 +0000 Subject: [PATCH 3/5] fix(reporting): run write triggers off the request path and bound platform reads Review follow-ups on the managed subscription path: - Reconciles triggered by served writes (sync_principal, sync_accounts, create_media_buy, update_media_buy) now run after the response on a background drain that coalesces per account and caller. A write's latency no longer depends on how many principals share the account, and the boundary sweep still catches anything missed. service.drain_subscription_triggers() waits for the drain; service.touch() stays awaited for hosts that want immediacy. The drain stops at shutdown. - SubscriptionOptions.read_timeout (30 seconds by default) bounds each read of the platform's list_accounts and get_media_buys. A hung read becomes SUBSCRIPTION_ROSTER_UNAVAILABLE for the roster. - Only write tasks an ADCPHandler application implements are wrapped. Wrapping the base not-supported default advertised tasks nobody served and conflicted with production ownership. - SubscriptionOptions.standard_dimensions supplies an offering's standard dimensions when its declaration cannot carry them yet, as production offerings validated against the served 3.3.0-beta.1 contract cannot. - The production plus managed composition is covered end to end, on memory and PostgreSQL: principal tasks delegated to the application, then sweep, planning and obligations. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reliable-reporting-service.md | 6 +- src/adcp/reporting/service.py | 23 +++- src/adcp/reporting/subscriptions/managed.py | 125 ++++++++++++++---- .../test_reporting_subscription_production.py | 108 ++++++++++++--- tests/test_reporting_subscription_managed.py | 45 +++++++ 5 files changed, 261 insertions(+), 46 deletions(-) diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index 05e6017f7..0ec8143c9 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -362,7 +362,11 @@ platform = reporting.install(MySeller()) # then serve the platform as usual `Account(timezone=..., operator=...)` serves both on a decisioning platform. - **Scheduling.** `sync_principal`, `sync_accounts`, `create_media_buy` and `update_media_buy` served through the installed platform reconcile the - principals concerned (`reconcile_on_write`, on by default). Each worker turn + principals concerned (`reconcile_on_write`, on by default). The reconcile + runs after the response, on a background drain, so a write never waits for + it (`await reporting.drain_subscription_triggers()` waits, for tests). Each + read of the platform is bounded by `read_timeout` (30 seconds by default). + Each worker turn first reconciles every principal a period boundary has passed for since its last reconcile, then plans. Call `await reporting.touch(account_id=...)` after a change the SDK did not serve, such as a seller UI edit. diff --git a/src/adcp/reporting/service.py b/src/adcp/reporting/service.py index 856a048df..20346d649 100644 --- a/src/adcp/reporting/service.py +++ b/src/adcp/reporting/service.py @@ -624,6 +624,10 @@ def __init__( ) except TypeError as error: raise ReliableReportingConfigurationError(str(error)) from None + owned_resources = ( + *owned_resources, + ReportingServiceResource(close=self._managed.aclose), + ) if reporting_subscriptions is not None and production is None: self._build_subscription_planner() self._lifecycle = _ServiceLifecycle( @@ -788,14 +792,23 @@ def _bind_handler(self, handler: Any) -> None: self._managed.bind_handler(handler) async def _subscription_write(self, account_id: str | None, context: Any | None) -> None: - """Trigger: a served account or media-buy write may change what is owed.""" + """Trigger: a served account or media-buy write may change what is owed. + + Queues the reconcile on the background drain, so the buyer's write + never waits for it. + """ if self._managed is None or not self._managed.composed: return try: - await self._managed.touch(account_id=account_id, context=context) + await self._managed.trigger(account_id=account_id, context=context) except Exception: # noqa: BLE001 - a trigger never fails the buyer's write logger.exception("Reliable Reporting subscription trigger failed") + async def drain_subscription_triggers(self) -> None: + """Wait until every reconcile queued by served writes has run (tests, shutdown).""" + if self._managed is not None: + await self._managed.drain() + def _compose_production(self) -> None: if self._production_options is None or self._production is not None: return @@ -1962,9 +1975,13 @@ async def sync_principal(instance: Any, params: Any, context: Any | None = None) "sync_principal", lambda: service.sync_principal(params, context) ) + from adcp.server import ADCPHandler + def triggered(name: str) -> Callable[..., Any] | None: method = getattr(original, name, None) - if method is None: + # Only tasks the application implements: wrapping the base + # not-supported default would advertise a task nobody serves. + if method is None or method is getattr(ADCPHandler, name, None): return None async def write(instance: Any, params: Any, context: Any | None = None) -> Any: diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py index 69df09bde..f211559d7 100644 --- a/src/adcp/reporting/subscriptions/managed.py +++ b/src/adcp/reporting/subscriptions/managed.py @@ -30,11 +30,12 @@ from __future__ import annotations +import asyncio import inspect import logging from collections.abc import Awaitable, Callable, Mapping, Sequence from dataclasses import dataclass, field -from datetime import datetime +from datetime import datetime, timedelta from typing import TYPE_CHECKING, Any from adcp._principal_ref import PrincipalRef @@ -137,10 +138,20 @@ class SubscriptionOptions: ``ToolContext.caller_identity``), with ``ToolContext.tenant_id``. ``reconcile_on_write`` (default on) reconciles a principal when a served - write changes what it is owed. ``principals``, ``principal_store`` and - ``state_store`` supply the principal runtime and the stores; share durable - stores across every worker in production. ``account_configuration`` - keeps ``sync_accounts`` in ``configuration_tasks``. + write changes what it is owed. The reconcile runs after the response, on + a background drain, so a write's latency never depends on how many + principals share the account; the worker's boundary sweep catches anything + it misses. ``read_timeout`` bounds each read of the platform's + ``list_accounts``/``get_media_buys`` (``None`` for no bound). + + ``principals``, ``principal_store`` and ``state_store`` supply the + principal runtime and the stores; share durable stores across every worker + in production. ``account_configuration`` keeps ``sync_accounts`` in + ``configuration_tasks``. ``standard_dimensions`` maps an ``offering_id`` + to its standard dimensions for an offering whose declaration cannot carry + them yet: production offerings validate against the served 3.3.0-beta.1 + contract, which lacks the field. It goes away once the SDK serves + 3.3.0-beta.3. """ max_accounts: int | None = None @@ -155,6 +166,8 @@ class SubscriptionOptions: is_period_delivered: PeriodDeliveredOverride | None = None resolve_principal: ManagedPrincipalResolver | None = None reconcile_on_write: bool = True + read_timeout: timedelta | None = timedelta(seconds=30) + standard_dimensions: Mapping[str, Sequence[str]] | None = None account_configuration: bool = True principal_kind: str = "buyer_agent" principals: PrincipalService | None = None @@ -170,6 +183,8 @@ def __post_init__(self) -> None: type(self.max_accounts) is not int or self.max_accounts < 1 ): raise ValueError("SubscriptionOptions.max_accounts must be a positive integer") + if self.read_timeout is not None and self.read_timeout <= timedelta(0): + raise ValueError("SubscriptionOptions.read_timeout must be positive") if self.principals is not None and self.principal_store is not None: raise ValueError("pass principals or principal_store, not both") if self.end_account_configs is not None and self.account_delivery_configs is None: @@ -424,6 +439,9 @@ def __init__( self._engine: ReportingSubscriptionEngine | None = None self._principals: PrincipalService | None = options.principals self._service_options: ReportingSubscriptionServiceOptions | None = None + #: 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 # -- composition ----------------------------------------------------------- @@ -464,7 +482,7 @@ def compose( from adcp.server.principal import PrincipalService options = self.options - mapped = self._offerings(offerings) + mapped = self._offerings(offerings, options.standard_dimensions or {}) dimensions = ( tuple(options.dimensions) if options.dimensions is not None @@ -511,7 +529,9 @@ def compose( return self._service_options @staticmethod - def _offerings(offerings: Sequence[Mapping[str, Any]]) -> list[SubscriptionOffering]: + def _offerings( + offerings: Sequence[Mapping[str, Any]], declared: Mapping[str, Sequence[str]] + ) -> list[SubscriptionOffering]: from adcp.reporting.subscriptions._clock import CoverageInputError from adcp.reporting.subscriptions.resolution import SubscriptionOffering @@ -524,6 +544,9 @@ def _offerings(offerings: Sequence[Mapping[str, Any]]) -> list[SubscriptionOffer ) mapped: list[SubscriptionOffering] = [] for offering in offerings: + offering_id = str(offering.get("offering_id")) + if "standard_dimensions" not in offering and offering_id in declared: + offering = {**offering, "standard_dimensions": list(declared[offering_id])} if "standard_dimensions" not in offering: raise ReliableReportingConfigurationError( f"offering {offering.get('offering_id')!r} has no standard_dimensions, so " @@ -647,12 +670,10 @@ async def sync_principal(self, params: Any, context: Any) -> Any: self._identity(principal), params, context=context ) result = getattr(response, "result", None) - if ( - self.options.reconcile_on_write - and getattr(result, "kind", None) == "applied" - and str(getattr(result, "action", "")) != "unchanged" + if getattr(result, "kind", None) == "applied" and ( + str(getattr(result, "action", "")) != "unchanged" ): - await self.reconcile(principal) + self._enqueue(None, principal) return response # -- scheduling ------------------------------------------------------------ @@ -674,21 +695,68 @@ async def reconcile(self, principal: PrincipalRef, *, now: datetime | None = Non return True async def touch(self, *, account_id: str | None, context: Any | None = None) -> None: - """Reconcile the principals a served write on ``account_id`` concerns. + """Reconcile now the principals a change on ``account_id`` concerns. - The caller, when it holds subscriptions, and every principal whose - engine state includes the account. + The caller ``context`` authenticates, when it holds subscriptions, and + every principal whose engine state includes the account. Awaited, for + hosts that want immediacy after a change the SDK did not serve. """ + if not self.composed: + return + caller = await self._caller(context) + await self._reconcile_concerned(account_id, caller) + + async def _caller(self, context: Any | None) -> PrincipalRef | None: + if context is None: + return None + try: + return await self.resolve(None, context) + except ReliableReportingConfigurationError: + return None + + async def trigger(self, *, account_id: str | None, context: Any | None = None) -> None: + """Queue a reconcile after a served write; returns without waiting for it.""" if not self.options.reconcile_on_write or not self.composed: return - concerned: dict[str, PrincipalRef] = {} - if context is not None: + self._enqueue(account_id, await self._caller(context)) + + def _enqueue(self, account_id: str | None, caller: PrincipalRef | None) -> None: + if not self.options.reconcile_on_write: + return + self._queued[(account_id, None if caller is None else caller.subject)] = caller + if self._drain is None or self._drain.done(): + self._drain = asyncio.get_running_loop().create_task( + self._run_drain(), name="adcp-reporting-subscription-triggers" + ) + + async def _run_drain(self) -> None: + while self._queued: + (account_id, _subject), caller = next(iter(self._queued.items())) + del self._queued[(account_id, _subject)] try: - caller = await self.resolve(None, context) - except ReliableReportingConfigurationError: - caller = None - if caller is not None: - concerned[caller.subject] = caller + await self._reconcile_concerned(account_id, caller) + except Exception: # noqa: BLE001 - the drain never dies on one trigger + logger.exception("Reliable Reporting subscription trigger failed") + + async def drain(self) -> None: + """Wait until every queued served-write trigger has reconciled.""" + while self._drain is not None and not self._drain.done(): + await asyncio.shield(self._drain) + + async def aclose(self) -> None: + """Stop the trigger drain; queued work is left to the boundary sweep.""" + self._queued.clear() + task, self._drain = self._drain, None + if task is not None and not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + async def _reconcile_concerned( + self, account_id: str | None, caller: PrincipalRef | None + ) -> None: + concerned: dict[str, PrincipalRef] = {} + if caller is not None: + concerned[caller.subject] = caller if account_id is not None: try: principals = await self.list_principals() @@ -737,7 +805,18 @@ async def _call(self, tool: str, request: Any, principal: PrincipalRef) -> dict[ "reporting subscriptions derive accounts and buys from the installed platform; " "call service.install(platform) first" ) - response = _wire(await _resolve(getattr(target, tool)(request, context))) + timeout = self.options.read_timeout + try: + response = _wire( + await asyncio.wait_for( + _resolve(getattr(target, tool)(request, context)), + None if timeout is None else timeout.total_seconds(), + ) + ) + except asyncio.TimeoutError: + raise RuntimeError( + f"{tool} did not answer within SubscriptionOptions.read_timeout" + ) from None if not isinstance(response, Mapping): raise TypeError(f"{tool} returned {type(response).__name__}") if response.get("supported") is False or response.get("errors"): diff --git a/tests/conformance/reporting/test_reporting_subscription_production.py b/tests/conformance/reporting/test_reporting_subscription_production.py index 39a523e1f..d3a6d218a 100644 --- a/tests/conformance/reporting/test_reporting_subscription_production.py +++ b/tests/conformance/reporting/test_reporting_subscription_production.py @@ -43,7 +43,10 @@ from adcp.reporting.receipts import ReportingReceiptError from adcp.reporting.service import ReliableReportingService, ReportingAccountContext from adcp.reporting.service_production import ReportingProductionOptions, ReportingServiceOffering -from adcp.reporting.service_subscriptions import ReportingSubscriptionServiceOptions +from adcp.reporting.service_subscriptions import ( + ReportingSubscriptionServiceOptions, + SubscriptionOptions, +) from adcp.reporting.subscriptions import ( AuthorizedAccount, ReportingSubscriptionEngine, @@ -152,16 +155,17 @@ async def _seller( *, pool_buys: tuple[str, ...] = ("mb-1",), notifications: bool = False, + managed: bool = False, ): async with AsyncExitStack() as stack: clock = DeterministicReportingClock(END) store = await _store(stack, backend, notifications=notifications) - async with _composed(tmp_path, clock, store, pool_buys, notifications) as h: + async with _composed(tmp_path, clock, store, pool_buys, notifications, managed) as h: yield h @asynccontextmanager -async def _composed(tmp_path, clock, store, pool_buys, notifications): +async def _composed(tmp_path, clock, store, pool_buys, notifications, managed=False): verifier = _subscription_verifier() key = verifier.key writer = SQLiteDestination(tmp_path / "destination.sqlite", key) @@ -308,27 +312,36 @@ async def authorize(account, context, consumer): ) failures: list[Any] = [] h.failures = failures - subscriptions = ReportingSubscriptionServiceOptions( - engine=engine, - list_principals=lambda: [AGENCY], - resolve_principal=lambda _request, _context: AGENCY, - bind_managed_generation=bind, - planning_error_handler=failures.append, - ) + composition: dict[str, Any] + if managed: + composition = { + "subscriptions": SubscriptionOptions( + accounts=host.list_authorized_accounts, + buys=host.account_buys, + resolve_principal=lambda _request, _context: AGENCY, + bind_managed_generation=bind, + planning_error_handler=failures.append, + # The served 3.3.0-beta.1 offering contract has no standard_dimensions. + standard_dimensions={OFFERING_ID: ["package"]}, + ) + } + else: + composition = { + "reporting_subscriptions": ReportingSubscriptionServiceOptions( + engine=engine, + list_principals=lambda: [AGENCY], + resolve_principal=lambda _request, _context: AGENCY, + bind_managed_generation=bind, + planning_error_handler=failures.append, + ) + } h.service = ( ReliableReportingService.memory( - account_context=context, - clock=clock, - production=options, - reporting_subscriptions=subscriptions, + account_context=context, clock=clock, production=options, **composition ) if store is None else ReliableReportingService( - store=store, - account_context=context, - clock=clock, - production=options, - reporting_subscriptions=subscriptions, + store=store, account_context=context, clock=clock, production=options, **composition ) ) h.service.sources.register( @@ -337,6 +350,9 @@ async def authorize(account, context, consumer): constituent_of=lambda row, request: request.coverage.constituents[0].constituent_id, ) handler = h.service.install(_App()) + h.handler = handler + if managed: + h.engine, h.principals = h.service._managed.engine, h.service.principals h.mount = create_mcp_server(handler) # production requires a mounted handler h.production = h.service._production h.store = h.service.store @@ -546,3 +562,57 @@ async def test_a_second_offering_of_one_version_is_refused(tmp_path, backend) -> assert h.failures[0].error.code == "reporting_production_offering_conflict" (key,) = h.bound assert await h.store.list_generation_obligations(key) == () + + +async def test_the_managed_path_serves_production_through_its_application( + tmp_path, backend +) -> None: + """``subscriptions=`` with production: principal tasks reach the application.""" + from adcp.server import ToolContext + from adcp.server.mcp_tools import get_tools_for_handler + + async with _seller(tmp_path, backend, managed=True) as h: + tools = {item["name"] for item in get_tools_for_handler(h.handler, _include_schemas=False)} + assert {"get_principal", "sync_principal"} <= tools + h.host.accounts[ACCOUNT] = AuthorizedAccount( + ACCOUNT, timezone="UTC", authorized_at=at("2026-08-01T00:00:00Z") + ) + h.host.buys[ACCOUNT] = [_buy("mb-1")] + h.clock.set(at("2026-08-31T22:30:00Z")) # accepted before the buy runs + response = await h.handler.sync_principal( + { + "idempotency_key": "managed-production-0001", + "configuration": { + "reporting_destinations": [DESTINATION], + "reporting_subscriptions": [ + { + "subscription_id": "analytics-hourly", + "subscription_version": 1, + "active": True, + "scope": {"all_authorized_accounts": True}, + "selector": { + "feed_purpose": "analytics", + "period_duration": "PT1H", + "required_finality": "snapshot", + "dimensions": ["package"], + }, + "reconciliation_mode": "delivery_only", + "destination": {"destination_id": "warehouse", "path_template": PATH}, + } + ], + }, + }, + ToolContext(), + ) + assert response.result.kind == "applied", response + await h.service.drain_subscription_triggers() + assert await h.engine.account_ids(AGENCY) == {ACCOUNT} + h.clock.set(END) + await h.service.start() # sweep, plan, produce + assert h.failures == [] + (key,) = h.bound + obligations = await h.store.list_generation_obligations(key) + assert [o.period.start for o in obligations] == [ + at("2026-08-31T23:00:00Z"), + at("2026-09-01T00:00:00Z"), + ] diff --git a/tests/test_reporting_subscription_managed.py b/tests/test_reporting_subscription_managed.py index 29af0cad6..6a71e2e69 100644 --- a/tests/test_reporting_subscription_managed.py +++ b/tests/test_reporting_subscription_managed.py @@ -144,6 +144,7 @@ async def subscribe(self, principal: str = AGENCY, **changes: Any) -> Any: _ctx(principal), ) assert response.result.kind == "applied", response + await self.service.drain_subscription_triggers() return response @@ -269,6 +270,7 @@ async def test_served_writes_reconcile_the_principals_concerned() -> None: await world.subscribe(OTHER) calls = len(world.seller.listings) await world.platform.create_media_buy({"account": {"account_id": ACCOUNT}}, _ctx()) + await world.service.drain_subscription_triggers() # Both principals hold state for the account. assert len(world.seller.listings) == calls + 2 await world.service.touch(account_id="unknown-account") @@ -287,6 +289,7 @@ async def test_unchanged_sync_principal_does_not_reconcile() -> None: _ctx(), ) assert response.result.action == "unchanged" + await world.service.drain_subscription_triggers() assert len(world.seller.listings) == 1 @@ -406,6 +409,7 @@ async def test_the_binding_keeps_no_credential_material() -> None: }, context, ) + await world.service.drain_subscription_triggers() binding = await world.service._managed.engine.binding(PrincipalRef(None, AGENCY)) assert binding == { "tenant_id": None, @@ -499,6 +503,7 @@ async def test_a_decisioning_platform_is_served_through_its_own_handler() -> Non context, ) assert response.result.kind == "applied" + await service.drain_subscription_triggers() # The roster was read through accounts.list under the restored identity. [listing] = platform.accounts.listed assert listing.auth_info.principal == AGENCY @@ -524,6 +529,46 @@ async def test_a_decisioning_platform_is_served_through_its_own_handler() -> Non ), ToolContext(metadata={"adcp.auth_info": {"kind": "bearer", "principal": AGENCY}}), ) + await service.drain_subscription_triggers() assert len(platform.accounts.listed) == 3 finally: current_principal.reset(token) + + +@pytest.mark.asyncio +async def test_a_served_write_never_waits_for_its_reconcile() -> None: + import asyncio + + world = _World() + await world.subscribe() + release = asyncio.Event() + listed = world.seller.list_accounts + + async def slow(params: Any, context: Any = None) -> Any: + await release.wait() + return await listed(params, context) + + world.seller.list_accounts = slow # type: ignore[method-assign] + result = await asyncio.wait_for( + world.platform.create_media_buy({"account": {"account_id": ACCOUNT}}, _ctx()), 1 + ) + assert result["media_buy_id"] == "media-buy-new" + release.set() + await world.service.drain_subscription_triggers() + await world.service.close() + + +@pytest.mark.asyncio +async def test_a_hung_platform_read_is_bounded_by_read_timeout() -> None: + import asyncio + + world = _World(read_timeout=timedelta(milliseconds=50)) + + async def hung(params: Any, context: Any = None) -> Any: + await asyncio.sleep(30) + + world.seller.list_accounts = hung # type: ignore[method-assign] + await world.subscribe() + [failure] = world.failures + assert failure.error.code == "SUBSCRIPTION_ROSTER_UNAVAILABLE" + assert "RuntimeError" in str(failure.error) From 8495b1788803e990764942ee5d0ff7ece4d0d508 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 20:47:38 +0000 Subject: [PATCH 4/5] fix(reporting): parse wire instants with 1-6 fractional digits on Python 3.10 Python 3.10's datetime.fromisoformat refuses fractional seconds with other than three or six digits. The managed path's account and buy instants, and the canonical-vector test's helper, now parse through the SDK's aware_timestamp. Full CI on Python 3.10 caught it on the fractional-seconds-and-offsets vector. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/managed.py | 6 +++++- tests/test_reporting_subscription_parity.py | 4 +++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py index f211559d7..36e55a16c 100644 --- a/src/adcp/reporting/subscriptions/managed.py +++ b/src/adcp/reporting/subscriptions/managed.py @@ -222,7 +222,11 @@ def _instant(value: Any) -> datetime | None: return None if isinstance(value, datetime): return utc_instant(value) - return utc_instant(datetime.fromisoformat(str(value).replace("Z", "+00:00"))) + from adcp.reporting._timestamp import aware_timestamp + + # Wire instants may carry 1-6 fractional digits, which Python 3.10's ISO + # parser refuses. + return aware_timestamp(str(value)) # -- principal identity -------------------------------------------------------- diff --git a/tests/test_reporting_subscription_parity.py b/tests/test_reporting_subscription_parity.py index cfe0c4938..1036353dc 100644 --- a/tests/test_reporting_subscription_parity.py +++ b/tests/test_reporting_subscription_parity.py @@ -25,6 +25,7 @@ import pytest +from adcp.reporting._timestamp import aware_timestamp from adcp.reporting.subscriptions import ( CoverageInputError, SubscriptionAuthorization, @@ -80,7 +81,8 @@ def pinned_tzdata(): def at(value: str | None) -> datetime | None: - return None if value is None else datetime.fromisoformat(value.replace("Z", "+00:00")) + # The vectors carry 1-6 fractional digits, which Python 3.10's ISO parser refuses. + return None if value is None else aware_timestamp(value) def windows( From d257b16e0422088d7de815e05e067cbddd752362 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 21:10:42 +0000 Subject: [PATCH 5/5] ci: pin pypa/gh-action-pypi-publish to v1.14.3 for GHSA-5xjx-3w67-w9x9 GHSA-5xjx-3w67-w9x9, published 2026-10-30, covers pypa/gh-action-pypi-publish up to 1.14.2: a revoked CA in the chain Sigstore uses for Trusted Publishing and PyPI attestations. zizmor's known-vulnerability audit now fails the Workflow security job on every branch. v1.14.3 is the first patched release, pinned by its commit. Co-Authored-By: Claude Opus 5.5 (1M context) --- .github/workflows/release-publish.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/release-publish.yml b/.github/workflows/release-publish.yml index a5ac8f1db..33e758298 100644 --- a/.github/workflows/release-publish.yml +++ b/.github/workflows/release-publish.yml @@ -115,6 +115,6 @@ jobs: with: name: release-dist path: dist/ - - uses: pypa/gh-action-pypi-publish@ed0c53931b1dc9bd32cbe73a98c7f6766f8a527e # v1.13.0 + - uses: pypa/gh-action-pypi-publish@7245906fb61fa33a13554f9d5fb856fe61758086 # v1.14.3 with: packages-dir: dist/