feat(reporting): serve reporting subscriptions from one SubscriptionOptions block

.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/

docs/reliable-reporting-service.md

@@ -324,6 +324,68 @@ 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). 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.
+- **`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 +408,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 (

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"]

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
 
 

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,
@@ -1320,6 +1322,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 +1531,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 +2923,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 +2983,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 +3073,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(  # 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(  # 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)
+
     async def get_reporting_status(  # type: ignore[override]
         self, params: GetReportingStatusRequest, context: ToolContext | None = None
     ) -> Any:
@@ -3266,7 +3313,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,

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

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",

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: