From 650d15584d8987e3441dd36eadfc843104f8d3f4 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sat, 31 Oct 2026 12:58:03 +0000 Subject: [PATCH 1/2] feat(reporting): apply the AdCP 3.3.0-beta.3 seller rules for subscriptions The seller engine follows the rest of adcp#8619 and #8620: - Admission under max_accounts orders candidates by authorized_at, then account_id in byte order, and keeps it sticky. A candidate with another open account-level exception still holds its place. - Several pacing or analytics subscriptions may apply to one account and purpose. Where two resolve the same offering for a package, the lowest subscription_id owes it and each other records a package-level overridden exception naming the package_ids. Coverage, on both sides, removes those packages while the entry is open. Billing stays one per account at a scope level. - Membership edits apply in place. Adding or removing account_ids and changing exclude_account_ids keep subscription_version and accepted_at. A removed account opens an excluded_by_buyer entry from the cutover and stays listed as excluded. An added account gets one closed entry for the time before it joined. The buyer's coverage reads the intervals rather than the current membership. - destination_overrides route an account to another destination. Coverage resolves against that destination's method, the engine routes its per-account resolution, and owed sets carry the override as the version's destination for the managed binding. Registration checks that every override names an existing destination, and that one routing several accounts uses account_id in its path_template. - The readback lists 16 prior versions and 1,000 accounts inline, with their totals. get_principal reporting_subscription_page pages the rest from the engine's state document, which now keeps the full lists. - AccountReady carries resolution, the matched or created account's money terms, effective_at for an updated money change, and not_carried_over. The buyer facade no longer bumps subscription_version for a membership edit. Co-Authored-By: Claude Opus 5.5 (1M context) --- MIGRATION_v9_to_v10.md | 1 + docs/adcp-3.3-sdk-inventory.md | 4 +- docs/reporting-subscriptions-buyer.md | 2 +- docs/reporting-subscriptions-seller.md | 10 + src/adcp/account_outcomes.py | 29 ++ src/adcp/accounts.py | 11 + src/adcp/reporting/subscriptions/coverage.py | 82 +++- src/adcp/reporting/subscriptions/engine.py | 410 ++++++++++++++++-- src/adcp/reporting/subscriptions/facade.py | 8 +- src/adcp/reporting/subscriptions/registry.py | 104 ++++- .../reporting/subscriptions/resolution.py | 30 ++ src/adcp/server/principal.py | 95 +++- tests/test_account_outcomes.py | 57 +++ .../test_principal_reporting_subscriptions.py | 117 ++++- ...est_reporting_subscription_beta3_engine.py | 285 ++++++++++++ 15 files changed, 1157 insertions(+), 88 deletions(-) create mode 100644 tests/test_reporting_subscription_beta3_engine.py diff --git a/MIGRATION_v9_to_v10.md b/MIGRATION_v9_to_v10.md index f6729d7c9..ab090ec6b 100644 --- a/MIGRATION_v9_to_v10.md +++ b/MIGRATION_v9_to_v10.md @@ -16,6 +16,7 @@ the SDK-relevant inventory is [docs/adcp-3.3-sdk-inventory.md](docs/adcp-3.3-sdk | AdCP 3.3.0-beta.3 composes the protocol envelope on every task response | adcontextprotocol/adcp#8547: every response root composes `core/protocol-envelope.json`, which composes `core/version-envelope.json`, so generated responses inherit `ProtocolEnvelope` alone (no separate `AdcpVersionEnvelope` base) and the local copies of envelope fields are gone. Submitted, working and input-required bodies compose `core/task-submitted.json` / `task-working.json` / `task-input-required.json`: a submitted arm still pins `status` to `submitted` and requires `task_id`. `task_id` (non-empty) and the task-level `message` (at most 2,000 characters) each have one definition (`core/task-id.json`, `core/task-message.json`); fields that refer to them stay typed `str` with those constraints, so no caller has to wrap a value. `TaskId`, `TaskMessage`, `TaskSubmitted`, `TaskWorking` and `TaskInputRequired` are generated. Consequences for handlers: `list_products` and `decline_proposals` now carry `status` (the server stamps `completed`; the 3.3 → 3.2 conversion still drops it for 3.2 callers); an absent `replayed` reads as `false` and `replayed: false` is valid; and on governance responses `status` is the task status, so a `check_governance` result names its decision in `verdict` and `report_plan_outcome` in `outcome_state`, never in `status`. `create_mcp_webhook_payload` gives a `result` without `status` the webhook's task status, and replaces a model result's defaulted envelope status with it, so a failed task's result never reads `completed`. | | AdCP 3.3.0-beta.3 adds live creative updates, account-request money terms and reporting-subscription fields | Experimental `creative.live_updates` (`LiveUpdatePolicy`, `CreativeStagedUpdate`, `creative.staged_update_changed` and `CreativeStagedUpdateChangedWebhook`, error code `CREATIVE_LIVE_UPDATE_UNSUPPORTED` with `CreativeLiveUpdateUnsupportedDetails`). `request_account` outcomes carry `account` money terms on `matched` / `created`, `effective_at` on an `updated` money change, and `not_carried_over`. Reporting subscriptions gain `destination_overrides[]`, the paged readback (`prior_versions_count`, `accounts_with_exceptions_count`, `accounts_unauthorized_count`, `get_principal` `reporting_subscription_page`), `authorization.regranted_at`, `flight_history[]` and `cancellation_effective_at` on media buys, `available_from` / `withdrawn_at` on offerings and `stub` on billing obligations. `comply_test_controller` adds the `reporting_subscription_probe` scenario: override `TestControllerStore.reporting_subscription_probe` to serve it. Generated renames: `PriorVersion` is `ReportingSubscriptionPriorVersion`, `AccountsWithException` is `ReportingSubscriptionAccountExceptions`, and `AccountsUnauthorizedItem` is `ReportingSubscriptionUnauthorizedAccount`. | | The reporting-subscription surface is served from 3.3.0-beta.3 on, unprojected | The beta.1 / beta.2 field projections are gone, with the preview module that patched the fields onto the models: `project_offering`, `project_reporting_delivery`, `project_subscription_state`, `project_subscription_coverage` and `serves_settled_subscription_fields` are removed from `adcp.reporting.subscriptions`, and `ReportingSubscriptionEngine.subscription_coverage()` no longer takes `adcp_version`. One predicate, `adcp.reporting.subscriptions.serves_subscriptions(adcp_version)` (from `SUBSCRIPTIONS_MIN_ADCP_VERSION`, `3.3.0-beta.3`), gates the whole surface: capabilities, `get_reporting_status` coverage (including `status: "inactive"` with `deactivated_at`) and the principal readback. A caller served an earlier release gets none of it. The extensible `reporting-dimension` vocabulary accepts unknown values on selectors, offerings and the capability block. | +| Reporting-subscription registration follows the 3.3.0-beta.3 rules | A caller may hold several pacing or analytics subscriptions per account and purpose (the lowest `subscription_id` owes a shared offering; the others record a package-level `overridden` exception), while billing stays one per account at a scope level and must use `match: exact` with `dimensions`. Membership edits (`scope.account_ids`, `exclude_account_ids`) no longer need a new `subscription_version`, and `client.reporting.subscribe` no longer bumps the version for them. `destination_overrides` must name existing destinations. The readback lists 16 prior versions and 1,000 accounts inline with `*_count` totals; page the rest with `get_principal` `reporting_subscription_page`. A partial grant refuses an unlisted reporting task with `SCOPE_INSUFFICIENT`. `AccountReady` gains `resolution`, `billing`, `billing_entity`, `payment_terms`, `effective_at` and `not_carried_over`. | | A server that restricts `supported_versions` runs on its newest listed release | Any server (`create_mcp_server`, `create_a2a_server`, `serve`, `MCPToolSet`, `ADCPAgentExecutor`, `create_tool_caller`, `PlatformHandler`, `create_adcp_server_from_platform`) with `supported_versions=["3.2"]` and no `adcp_version` runs its handlers on 3.2, instead of raising because the SDK default is outside the list. Pass `adcp_version` explicitly to choose another listed release. | | `server_version` pins keep wire precision | `ADCPClient(server_version=...)` and `get_server_version()` keep a 3.x pin at release precision: `"3.1.0-beta.1"` reads back as `"3.1-beta.1"` (SDK 9 reported the bundle key, `"3.1.0-beta.1"`). Stable pins still collapse (`"3.0.7"` → `"3.0"`). | | A server pin governs its handlers | Handlers run on one *application version*: the server's `adcp_version` pin, else the newest release in its `supported_versions`, else the packaged `3.3-beta.3`. Every dispatch path (MCP, A2A, `create_tool_caller`, the Decisioning dispatcher) converts a buyer's request to that release before validation, unknown-field policy and typed parsing, and converts the handler's response back to the buyer's release: up to `3.3-beta.2`, unchanged for a same-release buyer, or down to 3.1 / 3.0. `resolve_requested_adcp_version(params)` inside a handler returns the application version; `ToolContext.application_adcp_version` reports it directly, and `ToolContext.resolved_adcp_version` still reports the buyer's negotiated version. A 3.2-pinned seller's dict handlers therefore see exactly 3.2 payloads, its requests and responses are validated against the 3.2 schemas (with strict validation, a response its own 3.2 contract refuses fails before conversion), and a conversion the target schema refuses fails as `ADCPVersionConversionError` / `VERSION_UNSUPPORTED` instead of reaching the peer invalid. Typed handler parameters (`params: GetProductsRequest`) and typed responses still use the generated 3.3 classes: they are additive supersets of 3.2, so a 3.2 payload parses into them unchanged and dumps back as the same 3.2 payload, but a field 3.3 added reads as unset (its model default) on a 3.2 server. The unpinned default is unchanged: handlers receive `3.3-beta.2`. | diff --git a/docs/adcp-3.3-sdk-inventory.md b/docs/adcp-3.3-sdk-inventory.md index ed43efc43..7fd908226 100644 --- a/docs/adcp-3.3-sdk-inventory.md +++ b/docs/adcp-3.3-sdk-inventory.md @@ -281,8 +281,8 @@ in it. | Wire pin | `3.3.0-beta.3` / `3.3-beta.3`. `3.3-beta.2` is no longer advertised or accepted as a pin (as with `3.3-beta.0`, `3.3-beta.1` and the 3.2 candidates); its cache and compliance trees stay in the source tree and leave the wheel. | done | | Protocol envelope composition (adcontextprotocol/adcp#8547) | Every task response and async status payload composes `core/protocol-envelope.json`, which composes `core/version-envelope.json`; submitted, working and input-required bodies compose `core/task-submitted.json` / `task-working.json` / `task-input-required.json`; `task_id` and the task-level `message` have one definition each (`core/task-id.json`, `core/task-message.json`). | generated (the #1741 codegen layout). The response-arm emitter folds the composed task-state schema into each arm, so submitted arms keep `status: submitted` and a required `task_id`; `get_task_status`'s required `task_id` / `status` take the envelope's declarations instead of `Any`; `task_id` / `message` references generate as constrained `str` | | Live creative updates | Experimental `creative.live_updates`, `staged_update` on `list_creatives` / `sync_creatives`, the `creative.staged_update_changed` notification, `CREATIVE_LIVE_UPDATE_UNSUPPORTED`. | generated; no handler wiring (no new task) | -| Account requests | Money terms on `matched` / `created`, `effective_at` on `updated` money changes, `not_carried_over` on a separate account, consistent `currency` / `timezone` changeability. | generated; the `accounts.ensure` outcome mapping follows in the account slice | -| Reporting subscriptions (adcontextprotocol/adcp#8613, #8617, #8618, #8619, #8622) | `status: "inactive"` with `deactivated_at` on coverage; billing `stub` obligations and `cancellation_effective_at`; `flight_history[]`, `authorization.regranted_at`, `packages[].reporting_delivery_offering_ids`, offering `available_from` / `withdrawn_at`; sticky admission and per-package resolution, `destination_overrides[]`, the paged readback (`*_count`, `reporting_subscription_page`), in-place membership edits, several pacing / analytics feeds per purpose, `SCOPE_INSUFFICIENT` for partial grants, `account_id` in `primary_keys`; the day-start rule and the other period-arithmetic rulings. | generated. `_reporting_subscription_preview` and the beta.1 / beta.2 projections are removed; `serves_subscriptions` is the one version predicate. Engine conformance to the beta.3 rules and the re-pinned coverage vectors follow in the next slice | +| Account requests | Money terms on `matched` / `created`, `effective_at` on `updated` money changes, `not_carried_over` on a separate account, consistent `currency` / `timezone` changeability. | generated; `AccountReady` carries `resolution`, the money terms (`billing`, `billing_entity`, `payment_terms`), `effective_at` and `not_carried_over` | +| Reporting subscriptions (adcontextprotocol/adcp#8613, #8617, #8618, #8619, #8622) | `status: "inactive"` with `deactivated_at` on coverage; billing `stub` obligations and `cancellation_effective_at`; `flight_history[]`, `authorization.regranted_at`, `packages[].reporting_delivery_offering_ids`, offering `available_from` / `withdrawn_at`; sticky admission and per-package resolution, `destination_overrides[]`, the paged readback (`*_count`, `reporting_subscription_page`), in-place membership edits, several pacing / analytics feeds per purpose, `SCOPE_INSUFFICIENT` for partial grants, `account_id` in `primary_keys`; the day-start rule and the other period-arithmetic rulings. | generated. `_reporting_subscription_preview` and the beta.1 / beta.2 projections are removed; `serves_subscriptions` is the one version predicate. The coverage rule and the seller engine follow every listed rule, and the parity test runs the spec's 120 beta.3 vectors | | Extensible `reporting-dimension` | The schema marks the vocabulary `x-extensible`; the generator closes it. | `adcp.types._forward_compat` reopens the five fields that carry it to the vocabulary's `x-pattern` | | `comply_test_controller` `reporting_subscription_probe` | `revoke_grant`, `regrant`, `reduce_grant` (`allowed_tasks`), `set_account_timezone` (`timezone`), `advance_time` (`target_time`). | `TestControllerStore.reporting_subscription_probe` hook with parameter validation; not implemented by default | | Generated renames | `PriorVersion` → `ReportingSubscriptionPriorVersion`, `AccountsWithException` → `ReportingSubscriptionAccountExceptions`, `AccountsUnauthorizedItem` → `ReportingSubscriptionUnauthorizedAccount`; numbered names renumber (`Status20` → `Status54`, ...). `adcp.types._generated.AllowedTask` now resolves to the controller's `allowed_tasks` item (both are task-name tokens); `adcp.types.domains.core.AllowedTask` is unchanged. | generated | diff --git a/docs/reporting-subscriptions-buyer.md b/docs/reporting-subscriptions-buyer.md index 8d8649cc7..53b990f6b 100644 --- a/docs/reporting-subscriptions-buyer.md +++ b/docs/reporting-subscriptions-buyer.md @@ -21,7 +21,7 @@ result = await client.reporting.subscribe( result.action, result.subscription_version ``` -- **Versioning.** The version is bumped only when the selector, scope, destination, `reconciliation_mode`, `exclude_account_ids` or `operator_domains` change. +- **Versioning.** The version is bumped only when the selector, the scope kind, the destination or its `destination_overrides`, `reconciliation_mode` or `operator_domains` change. Adding or removing `account_ids`, and changing `exclude_account_ids`, apply in place (3.3.0-beta.3): the seller records each edit per account as an `excluded_by_buyer` interval, so periods that started before a removal stay owed and an added account is owed only from when it joined. - **Pre-dispatch checks.** These fail with `dispatched=False`: - the scope must be explicit; - the destination must exist; diff --git a/docs/reporting-subscriptions-seller.md b/docs/reporting-subscriptions-seller.md index 903a5730b..b440aa2e9 100644 --- a/docs/reporting-subscriptions-seller.md +++ b/docs/reporting-subscriptions-seller.md @@ -31,6 +31,16 @@ The SDK reads these surfaces under each principal's identity. A principal sees e A decisioning `Account` carries `timezone=` and `operator=`. An account without a timezone is refused rather than read as UTC. +## AdCP 3.3.0-beta.3 rules the engine applies + +- **Admission under `max_accounts`.** Candidates are ordered by `authorization.authorized_at`, then `account_id` in byte order, and an admitted account keeps its slot. A waiting candidate carries a `capacity` exception. +- **Several feeds per purpose.** A caller may hold several pacing or analytics subscriptions for one account. Where two resolve the same offering for a package, the lowest `subscription_id` owes it and the other records an `overridden` exception naming the `package_ids`. Billing stays one per account at a scope level. +- **Membership in place.** Membership edits (`account_ids`, `exclude_account_ids`) keep the version. A removed account stays listed, `excluded`, with an `excluded_by_buyer` entry from the cutover. An added account gets a closed entry covering the time before it joined. +- **`destination_overrides`.** An account named by an override, by `account_ids` or `operator_domains`, resolves against the override's destination method. Its owed sets carry the override as the version's `destination`, so `bind_managed_generation` binds and checks the override's `path_template`. +- **Bounded readback.** `prior_versions` lists the 16 newest inline, and the account lists 1,000, each with a total count. `get_principal` `reporting_subscription_page` pages the rest from the engine's state. +- **Partial grants.** A grant that keeps the account covered refuses a task it does not list with `SCOPE_INSUFFICIENT`. +- **Billing stubs and `regranted_at`.** A billing cutover owes a stub obligation (`stub: true`), and `list_accounts` serves `authorization.regranted_at`. + ## What the SDK derives - **The principal runtime.** `install()` serves `get_principal` and `sync_principal` and advertises the principal section. The service's `PrincipalService` is `reporting.principals`. diff --git a/src/adcp/account_outcomes.py b/src/adcp/account_outcomes.py index 12c313196..f4f4bf45a 100644 --- a/src/adcp/account_outcomes.py +++ b/src/adcp/account_outcomes.py @@ -90,6 +90,17 @@ class AccountReady: ``status`` is the account's status, which may still need buyer action before spend (``payment_required``, for example). + + On the ``request_account`` path (AdCP 3.3.0-beta.3, adcp#8620): + + * ``resolution`` is ``matched``, ``created`` or ``updated``. A matched or + created account carries its own money terms (:attr:`billing`, + :attr:`billing_entity`, :attr:`payment_terms`), which can differ from + the terms requested: read them before spending. + * ``effective_at`` is when changed money terms take effect on an updated + account; delivery before it is invoiced under the previous terms. + * ``not_carried_over`` names the account-level configurations a separate + new account (a change resolved as ``created``) does not have. """ account_id: str | None @@ -98,6 +109,24 @@ class AccountReady: request_id: str | None = None account: Mapping[str, Any] | None = None kind: Literal["ready"] = "ready" + resolution: str | None = None + effective_at: str | None = None + not_carried_over: tuple[str, ...] = () + + @property + def billing(self) -> Any: + """The account's ``billing`` (who is invoiced), as the seller resolved it.""" + return None if self.account is None else self.account.get("billing") + + @property + def billing_entity(self) -> Any: + """The account's ``billing_entity``, without write-only fields.""" + return None if self.account is None else self.account.get("billing_entity") + + @property + def payment_terms(self) -> Any: + """The account's ``payment_terms``, as the seller resolved them.""" + return None if self.account is None else self.account.get("payment_terms") @dataclass(frozen=True) diff --git a/src/adcp/accounts.py b/src/adcp/accounts.py index 271af2433..a79b72779 100644 --- a/src/adcp/accounts.py +++ b/src/adcp/accounts.py @@ -225,12 +225,23 @@ def _request_outcome( request_id = str(wire["request_id"]) if state == "resolved": account = wire.get("account") if isinstance(wire.get("account"), Mapping) else None + resolution = wire.get("resolution") + effective_at = wire.get("effective_at") return AccountReady( account_id=str(wire["account_id"]), status=str((account or {}).get("status") or "active"), via="request_account", request_id=request_id, account=account, + # adcp#8620: money terms ride on the resolved account; an updated + # money change says when it takes effect. + resolution=None + if resolution is None + else str(getattr(resolution, "value", resolution)), + effective_at=None if effective_at is None else str(effective_at), + not_carried_over=tuple( + str(getattr(item, "value", item)) for item in wire.get("not_carried_over") or () + ), ) if state == "in_review": return AccountPending( diff --git a/src/adcp/reporting/subscriptions/coverage.py b/src/adcp/reporting/subscriptions/coverage.py index 92db579fb..f92b470dd 100644 --- a/src/adcp/reporting/subscriptions/coverage.py +++ b/src/adcp/reporting/subscriptions/coverage.py @@ -385,13 +385,30 @@ class SubscriptionException: #: An entry for one offering's boundaries (``excluded[]`` or ``gaps[]`` #: ``offering_id``); ``None`` is every offering. offering_id: str | None = None + #: A package-level entry (``package_ids``, with ``media_buy_id``): it removes + #: those packages from the buy's coverage while open, never periods. An + #: ``overridden`` entry names the packages a lower ``subscription_id`` owes. + package_ids: tuple[str, ...] = () def __post_init__(self) -> None: object.__setattr__(self, "from_", utc_instant(self.from_, field="from")) if self.until is not None: object.__setattr__(self, "until", utc_instant(self.until, field="until")) + object.__setattr__(self, "package_ids", tuple(self.package_ids)) + + def removes_package(self, at: datetime, media_buy_id: str, package_id: str) -> bool: + """Whether this package-level entry takes the package out of coverage at ``at``.""" + return ( + bool(self.package_ids) + and package_id in self.package_ids + and (self.media_buy_id is None or self.media_buy_id == media_buy_id) + and self.from_ <= at + and (self.until is None or at < self.until) + ) def removes(self, start: datetime, media_buy_id: str, offering_id: str | None = None) -> bool: + if self.package_ids: + return False # removes packages from coverage, not periods if self.media_buy_id is not None and self.media_buy_id != media_buy_id: return False if self.offering_id is not None and offering_id is not None: @@ -405,8 +422,10 @@ def subscription_exceptions( ) -> tuple[SubscriptionException, ...]: """Normalize ``subscription_coverage`` ``excluded[]`` and ``gaps[]`` entries. - An ``offering_unsupported`` entry naming ``package_ids`` removes those - packages from the buy's coverage, not periods, so it is dropped here. + An entry naming ``package_ids`` (``offering_unsupported``, or + ``overridden`` by a lower ``subscription_id`` resolving the same offering, + adcp#8619) removes those packages from the buy's coverage while it is + open, not periods. """ result: list[SubscriptionException] = [] for entry in excluded: @@ -415,8 +434,6 @@ def subscription_exceptions( continue if hasattr(entry, "model_dump"): entry = entry.model_dump(mode="json", exclude_none=True, by_alias=True) - if _value(entry, "package_ids"): - continue reason = _text(_value(entry, "reason")) if reason not in _REASONS: raise CoverageInputError( @@ -434,6 +451,7 @@ def subscription_exceptions( # "all" across offerings with different boundaries) removes # only that offering's periods. offering_id=None if offering_id is None else _text(offering_id), + package_ids=tuple(_text(item) for item in _value(entry, "package_ids") or ()), ) ) for gap in gaps: @@ -588,25 +606,34 @@ def subscription_version_windows( def subscription_covers_account( - configuration: Any, *, account_id: str, account_operator: str | None = None + configuration: Any, + *, + account_id: str, + account_operator: str | None = None, + membership_intervals: bool = False, ) -> bool: """Whether a subscription version's scope selects the account. ``exclude_account_ids`` and ``operator_domains`` narrow the scope; the seller also reports an excluded account as ``excluded_by_buyer``. With ``operator_domains``, an account whose operator is unknown is not selected. + + ``membership_intervals``: the account's ``excluded_by_buyer`` entries + record its membership edits (adcp#8619), which apply in place within a + version, so ``account_ids`` and ``exclude_account_ids`` do not decide it; + the entries remove the periods outside its membership. """ if hasattr(configuration, "model_dump"): configuration = configuration.model_dump(mode="json", exclude_none=True) scope = _value(configuration, "scope") or {} account_ids = _value(scope, "account_ids") if account_ids is not None: - if account_id not in {_text(item) for item in account_ids}: + if not membership_intervals and account_id not in {_text(item) for item in account_ids}: return False elif _value(scope, "all_authorized_accounts") is not True: return False excluded = _value(configuration, "exclude_account_ids") or () - if account_id in {_text(item) for item in excluded}: + if not membership_intervals and account_id in {_text(item) for item in excluded}: return False domains = _value(configuration, "operator_domains") if domains: @@ -961,7 +988,10 @@ def owed_subscription_periods( ) require_zone(change.previous_timezone, field="timezone_cutovers.previous_timezone") changes.sort(key=lambda item: item.changed_at) - removals = subscription_exceptions(excluded, gap_entries) + every_removal = subscription_exceptions(excluded, gap_entries) + removals = tuple(item for item in every_removal if not item.package_ids) + package_removals = tuple(item for item in every_removal if item.package_ids) + membership_intervals = any(item.reason == "excluded_by_buyer" for item in every_removal) ids = [item.offering_id for item in offerings] if len(set(ids)) != len(ids): raise CoverageInputError("invalid_input", "offering_id values must be unique") @@ -984,6 +1014,10 @@ def owed_subscription_periods( if item.withdrawn_at is not None: points.add(item.withdrawn_at) points.update(change.changed_at for change in changes) + for removal in package_removals: + points.add(removal.from_) + if removal.until is not None: + points.add(removal.until) ordered = sorted(points) selectors = { window.subscription_version: SubscriptionSelector.from_subscription(window.configuration) @@ -991,7 +1025,10 @@ def owed_subscription_periods( } in_scope = { window.subscription_version: subscription_covers_account( - window.configuration, account_id=account_id, account_operator=account_operator + window.configuration, + account_id=account_id, + account_operator=account_operator, + membership_intervals=membership_intervals, ) for window in windows } @@ -1011,6 +1048,9 @@ def owed_subscription_periods( destinations, account_timezone, changes, + account_id=account_id, + account_operator=account_operator, + package_removals=package_removals, ) flights_start = buy._earliest_start() latest_stop = buy._latest_stop() @@ -1132,6 +1172,10 @@ def _regimes( destinations: Mapping[str, Any] | None, account_timezone: str | None, changes: Sequence[AccountTimezoneChange], + *, + account_id: str = "", + account_operator: str | None = None, + package_removals: Sequence[SubscriptionException] = (), ) -> list[_Regime | None]: """Maximal stretches with one active, in-scope version and one resolution.""" regimes: list[_Regime | None] = [] @@ -1143,22 +1187,30 @@ def _regimes( regimes.append(None) continue selector = selectors[window.subscription_version] - destination = window.destination - if destination is None and selector.destination_id is not None: - if destinations is None or selector.destination_id not in destinations: + # adcp#8619: an account routed by destination_overrides resolves + # against its own destination's method. + destination_id = selector.destination_for(account_id, account_operator) + destination = window.destination if destination_id == selector.destination_id else None + if destination is None and destination_id is not None: + if destinations is None or destination_id not in destinations: raise CoverageInputError( "missing_destination_method", f"version {window.subscription_version} names destination " - f"{selector.destination_id!r}; pass its pattern and transport", + f"{destination_id!r}; pass its pattern and transport", ) - destination = destinations[selector.destination_id] + destination = destinations[destination_id] # adcp#8619: each package resolves when its (buy, version) pair first # does and keeps that result; a buy owes the union of its packages'. first_resolved = window.accepted_at if buy.confirmed_at is not None: first_resolved = max(first_resolved, buy.confirmed_at) found: dict[str, SubscriptionOffering] = {} - for _package_id, offering_ids in buy.product_offering_ids(): + for package_id, offering_ids in buy.product_offering_ids(): + if any( + item.removes_package(start, buy.media_buy_id, package_id) + for item in package_removals + ): + continue for item in resolve_reporting_subscription_package( selector, offerings, diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index e4ddf7024..a45d89b05 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -24,12 +24,25 @@ :func:`~adcp.reporting.subscriptions.owed_subscription_periods` a buyer runs, so both sides agree period by period. -``max_accounts`` order - Among accounts eligible to be served (in scope, authorized, not excluded - or overridden), accounts already served keep their slots; the remaining - slots go to the others in ``account_id`` byte order. Accounts beyond the - limit get a ``capacity`` exception. A newly granted account never - displaces one being served. This matches the TypeScript SDK. +``max_accounts`` order (adcp#8619) + Candidates are the accounts the scope selects that the caller is + authorized for, whether or not they have covered buys; one with another + open account-level exception still holds its place. Accounts already + admitted keep their slots; the remaining slots go to the others ordered + by ``authorization.authorized_at``, then ``account_id`` in byte order, the + same key for every subscription of a principal. Accounts beyond the limit + get a ``capacity`` exception. A newly granted account never displaces one + being served. + +Other 3.3.0-beta.3 rules (adcp#8619) + Several pacing or analytics subscriptions may apply to one account and + purpose: where two resolve the same offering for a package, the lowest + ``subscription_id`` owes it and each other records a package-level + ``overridden`` exception. Membership edits (``account_ids``, + ``exclude_account_ids``) apply in place as ``excluded_by_buyer`` + intervals. ``destination_overrides`` route an account to another + destination. The readback lists 16 prior versions and 1000 accounts + inline, with totals, and pages the rest through ``get_principal``. """ from __future__ import annotations @@ -324,6 +337,12 @@ class _PrincipalState: #: 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) + #: subscription_id -> every prior version seen, newest first; the readback + #: lists only the 16 newest inline (adcp#8619). + prior_versions: dict[str, list[dict[str, Any]]] = field(default_factory=dict) + #: subscription_id -> the full accounts_with_exceptions and + #: accounts_unauthorized lists last evaluated; the readback caps them at 1000. + readback: dict[str, dict[str, list[dict[str, Any]]]] = field(default_factory=dict) #: Version of the persisted engine-state document; bump on a breaking change. @@ -607,6 +626,25 @@ def encode_principal_state( if principal.replacements else {} ), + **( + { + "prior_versions": { + sid: _json_value(items) + for sid, items in sorted(principal.prior_versions.items()) + } + } + if principal.prior_versions + else {} + ), + **( + { + "readback": { + sid: _json_value(lists) for sid, lists in sorted(principal.readback.items()) + } + } + if principal.readback + else {} + ), } @@ -682,6 +720,12 @@ def decode_principal_state(document: Mapping[str, Any]) -> _PrincipalState: 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"])) + for sid, items in (document.get("prior_versions") or {}).items(): + principal.prior_versions[sid] = [dict(item) for item in items] + for sid, lists in (document.get("readback") or {}).items(): + principal.readback[sid] = { + name: [dict(item) for item in value] for name, value in lists.items() + } return principal @@ -798,6 +842,28 @@ def principal_options( record_versions=self.record_versions, clock=self._clock, replace_account_configs=self._end_account_configs is not None, + subscription_list=self.subscription_list, + ) + + async def subscription_list( + self, identity: PrincipalLike, subscription_id: str, list_name: str + ) -> list[dict[str, Any]] | None: + """The full readback list behind ``reporting_subscription_page`` (adcp#8619). + + ``prior_versions`` newest first, the account lists in ``account_id`` + byte order. ``None`` when the engine holds nothing for it, so the + inline list is paged instead. + """ + principal = await self._read(identity.subject) + if list_name == "prior_versions": + known = principal.prior_versions.get(subscription_id) + return None if known is None else [dict(item) for item in known] + lists = principal.readback.get(subscription_id) + if lists is None or list_name not in lists: + return None + return sorted( + (dict(item) for item in lists[list_name]), + key=lambda item: str(item.get("account_id")).encode("utf-8"), ) @property @@ -1061,6 +1127,18 @@ def change(principal: _PrincipalState) -> None: key = (str(payload["subscription_id"]), int(payload["subscription_version"])) configuration = dict(payload["configuration"]) principal.versions.setdefault(key, configuration) + inline = payload.get("prior_versions") or () + if inline: + # Every version passes through the inline list, newest + # first, so the full history accumulates here. + known = principal.prior_versions.setdefault(key[0], []) + seen = {int(item["subscription_version"]) for item in known} + known.extend( + _json_value(item) + for item in inline + if int(item["subscription_version"]) not in seen + ) + known.sort(key=lambda item: -int(item["subscription_version"])) destination = configuration.get("destination") if destination is not None and key not in principal.destinations: target = destinations.get(destination["destination_id"]) @@ -1437,6 +1515,14 @@ def _evaluate_subscription( now: datetime, ) -> tuple[str, list[dict[str, Any]], list[dict[str, Any]]]: subscription_id = str(state["subscription_id"]) + accepted = principal.versions.get( + (subscription_id, int(state["subscription_version"])), state["configuration"] + ) + history = { + account_id: entries + for (sub_id, account_id), entries in principal.exclusions.items() + if sub_id == subscription_id + } desired = self._desired_exclusions( state, states, @@ -1446,17 +1532,36 @@ def _evaluate_subscription( destinations, now, served=principal.served.setdefault(subscription_id, set()), + accepted=accepted, + history=history, ) covered = principal.covered.setdefault(subscription_id, set()) for account_id, wanted in desired.items(): + routed = _routed(state["configuration"], account_id, accounts[account_id], destinations) + if _joined_in_place(state["configuration"], accepted, account_id, history): + # adcp#8619: an added account with no entry is not owed the + # periods before it joined. + principal.exclusions.setdefault((subscription_id, account_id), []).append( + _Exclusion( + "excluded_by_buyer", + self._cutover( + state, + accounts[account_id], + buys.get(account_id, ()), + routed, + _instant_of(state["accepted_at"]), + ), + self._cutover( + state, accounts[account_id], buys.get(account_id, ()), routed, now + ), + ) + ) # Covered: served, with no account-level exception. if not any(item.media_buy_id is None for item in wanted): covered.add(account_id) # Surface an account the schedules cannot be derived for (no # timezone yet) now, rather than at the first owed_periods. - self._schedules( - state, accounts[account_id], buys.get(account_id, ()), destinations, now - ) + self._schedules(state, accounts[account_id], buys.get(account_id, ()), routed, now) self._sync_exclusions( principal, subscription_id, @@ -1465,7 +1570,7 @@ def _evaluate_subscription( state, accounts, buys.get(account_id, ()), - destinations, + routed, now, ) # Exceptions close only when their cause ends. While the caller has @@ -1482,11 +1587,18 @@ def _evaluate_subscription( entry.until = self._cutover( state, account, buys.get(account_id, ()), destinations, now ) - return ( - subscription_id, - self._accounts_with_exceptions(principal, subscription_id, now), - self._accounts_unauthorized(principal, state, accounts, buys, destinations, now), + with_exceptions = self._accounts_with_exceptions(principal, subscription_id, now) + unauthorized = self._accounts_unauthorized( + principal, state, accounts, buys, destinations, now ) + principal.readback[subscription_id] = { + "accounts_with_exceptions": [_json_value(item) for item in with_exceptions], + "accounts_unauthorized": [ + {key: _json_value(value) for key, value in item.items() if key != "_ended"} + for item in unauthorized + ], + } + return (subscription_id, with_exceptions, unauthorized) async def _destinations(self, identity: PrincipalLike) -> dict[str, Any]: return { @@ -1505,11 +1617,15 @@ def _desired_exclusions( now: datetime, *, served: set[str], + accepted: Mapping[str, Any] | None = None, + history: Mapping[str, Sequence[_Exclusion]] | None = None, ) -> dict[str, list[_Exclusion]]: """Per account the scope includes: the exclusions that should be open now. ``served`` holds the accounts with a ``max_accounts`` slot and is - updated in place. + updated in place. ``accepted`` is the version's configuration as + accepted and ``history`` the subscription's exception entries per + account, from which membership edits are read (adcp#8619). """ result: dict[str, list[_Exclusion]] = {} if not _active(state): @@ -1518,7 +1634,10 @@ def _desired_exclusions( scope = configuration["scope"] feed_purpose = configuration["selector"]["feed_purpose"] excluded_ids = set(configuration.get("exclude_account_ids") or ()) - eligible: list[tuple[int, bytes, str]] = [] + accepted = configuration if accepted is None else accepted + history = history or {} + eligible: list[tuple[int, datetime, bytes, str]] = [] + overridden: dict[str, _Exclusion] = {} for account_id, account in accounts.items(): if not account.authorized: continue @@ -1527,6 +1646,10 @@ def _desired_exclusions( result[account_id] = [_Exclusion("excluded_by_buyer", now)] continue if not _scope_selects(configuration, account_id, account): + if _left_in_place(configuration, accepted, account_id, account, history): + # adcp#8619: a removed account opens an excluded_by_buyer + # entry at the cutover; earlier periods stay owed. + result[account_id] = [_Exclusion("excluded_by_buyer", now)] continue overriding = _override( configuration, @@ -1539,35 +1662,152 @@ def _desired_exclusions( ) if overriding is not None: kind, identifier, effective_at = overriding - result[account_id] = [ - _Exclusion( - "overridden", - now, - overridden_by=(kind, identifier), - effective_at=effective_at, - ) - ] - continue - result[account_id] = [] + overridden[account_id] = _Exclusion( + "overridden", + now, + overridden_by=(kind, identifier), + effective_at=effective_at, + ) + # adcp#8619 admission: candidates hold their place whatever their + # other account-level exceptions, ordered by authorized_at, then + # account_id in byte order; an admitted account keeps its slot. + start = self._chain_start(account) or now eligible.append( - (0 if account_id in served else 1, account_id.encode("utf-8"), account_id) + (0 if account_id in served else 1, start, account_id.encode("utf-8"), account_id) ) limit = self._support.max_accounts - ranked = [account_id for _held, _key, account_id in sorted(eligible)] + ranked = [item[-1] for item in sorted(eligible)] kept = ranked if limit is None else ranked[:limit] served.clear() served.update(kept) - for account_id in ranked[len(kept) :]: - result[account_id] = [_Exclusion("capacity", now)] + admitted = set(kept) + for account_id in ranked: + entries: list[_Exclusion] = [] + if account_id in overridden: + entries.append(overridden[account_id]) + if account_id not in admitted: + entries.append(_Exclusion("capacity", now)) + result[account_id] = entries for account_id, entries in result.items(): if entries: continue account = accounts[account_id] + routed = _routed(configuration, account_id, account, destinations) + entries.extend(self._unsupported(state, account, buys.get(account_id, ()), routed, now)) entries.extend( - self._unsupported(state, account, buys.get(account_id, ()), destinations, now) + self._package_overrides( + state, + states, + account_id, + account, + buys.get(account_id, ()), + destinations, + configs.get(account_id, ()), + now, + ) ) return result + def _package_overrides( + self, + state: Mapping[str, Any], + states: Sequence[Mapping[str, Any]], + account_id: str, + account: _Account, + buys: Sequence[SubscriptionBuy], + destinations: Mapping[str, Any], + configs: Sequence[AccountDeliveryConfig], + now: datetime, + ) -> list[_Exclusion]: + """Packages a lower ``subscription_id`` of the same purpose owes (adcp#8619). + + Pacing and analytics allow several subscriptions per account and + purpose. Where two resolve the same offering for the same package, the + lowest ``subscription_id`` in byte order owes it, and each other + records an ``overridden`` exception for that buy naming the packages. + """ + configuration = state["configuration"] + feed_purpose = configuration["selector"]["feed_purpose"] + if feed_purpose == "billing": + return [] + own_id = str(state["subscription_id"]).encode("utf-8") + level = "account_ids" if "account_ids" in configuration["scope"] else "all" + rivals = [] + for other in states: + other_config = other["configuration"] + other_id = str(other["subscription_id"]) + if ( + other_id.encode("utf-8") >= own_id + or not _active(other) + or other_config["selector"]["feed_purpose"] != feed_purpose + or ("account_ids" if "account_ids" in other_config["scope"] else "all") != level + or not _scope_selects(other_config, account_id, account) + or _override( + other_config, + other_config["scope"], + feed_purpose, + account_id, + account, + states, + configs, + ) + is not None + ): + continue + rivals.append((other_id, other_config)) + if not rivals: + return [] + rivals.sort(key=lambda item: item[0].encode("utf-8")) + own_selector = SubscriptionSelector.from_subscription(configuration) + own_destination = _destination_for( + own_selector, _routed(configuration, account_id, account, destinations) + ) + entries: list[_Exclusion] = [] + for buy in sorted(buys, key=lambda item: item.media_buy_id.encode("utf-8")): + if buy.confirmed_at is None: + continue + owed_by: dict[str, list[str]] = {} + for package_id, offering_ids in buy.product_offering_ids(): + if not package_id: + continue # an exception names packages; a packageless buy has none + mine = set( + resolve_subscription_offerings( + own_selector, + self._offerings, + product_offering_ids=offering_ids, + destination=own_destination, + at=now, + ).offering_ids + ) + if not mine: + continue + for other_id, other_config in rivals: + other_selector = SubscriptionSelector.from_subscription(other_config) + theirs = resolve_subscription_offerings( + other_selector, + self._offerings, + product_offering_ids=offering_ids, + destination=_destination_for( + other_selector, + _routed(other_config, account_id, account, destinations), + ), + at=now, + ).offering_ids + if mine & set(theirs): + owed_by.setdefault(other_id, []).append(package_id) + break + for other_id, package_ids in sorted(owed_by.items()): + entries.append( + _Exclusion( + "overridden", + now, + media_buy_id=buy.media_buy_id, + package_ids=tuple(sorted(package_ids)), + overridden_by=("subscription_id", other_id), + ) + ) + return entries + def _unsupported( self, state: Mapping[str, Any], @@ -1782,9 +2022,12 @@ async def subscription_coverage( result = [] for state in states: configuration = state["configuration"] - if not _scope_includes(configuration, account_id, account): - continue entries = principal.exclusions.get((str(state["subscription_id"]), account_id), []) + # adcp#8619: a removed account stays listed, excluded, while its + # open excluded_by_buyer entry is retained. + removed = any(e.reason == "excluded_by_buyer" and e.until is None for e in entries) + if not removed and not _scope_includes(configuration, account_id, account): + continue account_level = [e for e in entries if e.until is None and e.media_buy_id is None] status = "active" if any(e.reason == "action_required" for e in account_level): @@ -1978,9 +2221,10 @@ def _owed( if sub_id == subscription_id } exclusions = [ - SubscriptionException(e.reason, e.from_, e.until, e.media_buy_id) + SubscriptionException( + e.reason, e.from_, e.until, e.media_buy_id, package_ids=e.package_ids + ) for e in principal.exclusions.get((subscription_id, account_id), []) - if not e.package_ids ] # The seller never guesses a prior version's configuration, and each # version resolves against the destination it named when accepted. @@ -2061,7 +2305,10 @@ async def frozen_owed_sets( subscription_id=subscription_id, through=instant, periods=tuple(periods), - versions=versions, + versions={ + version: _routed_version(configuration, key, account) + for version, configuration in versions.items() + }, ) ) return result @@ -2172,6 +2419,97 @@ def _scope_selects(configuration: Mapping[str, Any], account_id: str, account: _ return False # operator_domains with an account the host gave no operator +def _routed( + configuration: Mapping[str, Any], + account_id: str, + account: _Account, + destinations: Mapping[str, Any], +) -> Mapping[str, Any]: + """``destinations`` with the subscription's destination routed for one account. + + ``destination_overrides`` (adcp#8619) send an account's data to another + destination; mapping it under the subscription's ``destination_id`` lets + every per-account resolution read it unchanged. + """ + destination = configuration.get("destination") or {} + default = destination.get("destination_id") + target = SubscriptionSelector.from_subscription(configuration).destination_for( + account_id, account.operator + ) + if default is None or target is None or target == default or target not in destinations: + return destinations + return {**destinations, default: destinations[target]} + + +def _routed_version( + configuration: Mapping[str, Any], account_id: str, account: _Account +) -> Mapping[str, Any]: + """A version's configuration with ``destination`` as routed for one account. + + The managed binding then resolves and checks the override's + ``destination_id`` and ``path_template`` (adcp#8619). + """ + for override in configuration.get("destination_overrides") or (): + account_ids = override.get("account_ids") + if account_ids is not None and account_id in account_ids: + break + else: + override = next( + ( + item + for item in configuration.get("destination_overrides") or () + if item.get("operator_domains") and account.operator in item["operator_domains"] + ), + None, + ) + if override is None: + return configuration + destination = {"destination_id": override["destination_id"]} + if override.get("path_template") is not None: + destination["path_template"] = override["path_template"] + return {**configuration, "destination": destination} + + +def _member_ids(configuration: Mapping[str, Any]) -> set[str] | None: + account_ids = (configuration.get("scope") or {}).get("account_ids") + return None if account_ids is None else set(account_ids) + + +def _left_in_place( + configuration: Mapping[str, Any], + accepted: Mapping[str, Any], + account_id: str, + account: _Account, + history: Mapping[str, Sequence[_Exclusion]], +) -> bool: + """Whether an ``account_ids`` edit removed the account within this version.""" + members = _member_ids(configuration) + if members is None or account_id in members or not _operator_selected(configuration, account): + return False + was_member = account_id in (_member_ids(accepted) or set()) + edited = any(item.reason == "excluded_by_buyer" for item in history.get(account_id, ())) + return was_member or edited + + +def _joined_in_place( + configuration: Mapping[str, Any], + accepted: Mapping[str, Any], + account_id: str, + history: Mapping[str, Sequence[_Exclusion]], +) -> bool: + """Whether an ``account_ids`` edit added the account within this version, unrecorded.""" + members = _member_ids(configuration) + if members is None or account_id not in members: + return False + return account_id not in (_member_ids(accepted) or set()) and not history.get(account_id) + + +def _instant_of(value: Any) -> datetime: + if isinstance(value, datetime): + return utc_instant(value) + return utc_instant(datetime.fromisoformat(str(value).replace("Z", "+00:00"))) + + def _scope_includes(configuration: Mapping[str, Any], account_id: str, account: _Account) -> bool: """Whether the scope names the account, counting buyer exclusions as included.""" excluded = set(configuration.get("exclude_account_ids") or ()) diff --git a/src/adcp/reporting/subscriptions/facade.py b/src/adcp/reporting/subscriptions/facade.py index 1fc63950d..5c504c8a1 100644 --- a/src/adcp/reporting/subscriptions/facade.py +++ b/src/adcp/reporting/subscriptions/facade.py @@ -286,9 +286,11 @@ async def subscribe( """Create or update one subscription, keeping every other entry unchanged. ``subscription_version`` is optional: the SDK keeps it when only - ``active`` or ``replace_account_configs`` change, and bumps it when - the selector, scope, destination, ``reconciliation_mode``, - ``exclude_account_ids`` or ``operator_domains`` change. A supplied + ``active``, ``replace_account_configs`` or the membership + (``scope.account_ids``, ``exclude_account_ids``) change, which apply + in place (adcp#8619), and bumps it when the selector, the scope kind, + the destination or its overrides, ``reconciliation_mode`` or + ``operator_domains`` change. A supplied version must be greater than the current one when the content changed. ``register_destination`` upserts a destination in the same atomic write. diff --git a/src/adcp/reporting/subscriptions/registry.py b/src/adcp/reporting/subscriptions/registry.py index c54bc7c73..27c18d882 100644 --- a/src/adcp/reporting/subscriptions/registry.py +++ b/src/adcp/reporting/subscriptions/registry.py @@ -21,12 +21,17 @@ ``file_transfer``), and shared templates differ; * ``subscription_id`` values are unique and never equal a ``delivery_config_id`` the caller uses through ``sync_accounts``; -* two active subscriptions at the same scope level may not both apply to one - account and ``feed_purpose``. +* two active billing subscriptions at the same scope level may not both apply + to one account; pacing and analytics allow several (adcp#8619); +* every ``destination_overrides[].destination_id`` names a destination that + exists after the same request applies, and overrides follow the + ``path_template`` rules of the destinations they route to. Versioning: a ``subscription_version`` is immutable. Resending it with -different semantic content is rejected; ``active`` and -``replace_account_configs`` change in place. A higher version records its own +different semantic content is rejected; ``active``, +``replace_account_configs`` and the membership (``scope.account_ids`` and +``exclude_account_ids``, adcp#8619) change in place. The engine records each +membership edit per account as ``excluded_by_buyer`` intervals. A higher version records its own ``accepted_at`` and moves the old one to ``prior_versions`` with ``superseded_at``. Omitting a subscription, or ``active: false``, deactivates it at ``deactivated_at``; it stays readable as ``inactive``. @@ -79,11 +84,27 @@ def _semantic(configuration: Mapping[str, Any]) -> dict[str, Any]: payload = {key: value for key, value in configuration.items() if key not in _IN_PLACE_FIELDS} scope = dict(payload.get("scope") or {}) if "account_ids" in scope: - scope["account_ids"] = sorted(scope["account_ids"]) + # adcp#8619: membership edits apply in place; only the scope kind counts. + scope["account_ids"] = [] payload["scope"] = scope - for name in ("exclude_account_ids", "operator_domains"): - if name in payload: - payload[name] = sorted(payload[name]) + payload.pop("exclude_account_ids", None) + if "operator_domains" in payload: + payload["operator_domains"] = sorted(payload["operator_domains"]) + if "destination_overrides" in payload: + payload["destination_overrides"] = sorted( + ( + { + **item, + **{ + name: sorted(item[name]) + for name in ("account_ids", "operator_domains") + if name in item + }, + } + for item in payload["destination_overrides"] + ), + key=lambda item: (item["destination_id"], str(item)), + ) selector = dict(payload.get("selector") or {}) if "dimensions" in selector: selector["dimensions"] = sorted(selector["dimensions"]) @@ -174,12 +195,32 @@ def _validate_one( raise ReportingSubscriptionValidationError( f"{where}: consumer_receipt requires a destination" ) + for override_index, override in enumerate(configuration.get("destination_overrides") or ()): + _validate_destination( + override, + f"{where}.destination_overrides[{override_index}]", + configuration, + destinations, + ) if destination is None: + if configuration.get("destination_overrides"): + raise ReportingSubscriptionValidationError( + f"{where}.destination_overrides needs a destination for the other accounts" + ) return + _validate_destination(destination, f"{where}.destination", configuration, destinations) + + +def _validate_destination( + destination: Mapping[str, Any], + where: str, + configuration: Mapping[str, Any], + destinations: Mapping[str, Mapping[str, Any]], +) -> None: target = destinations.get(destination["destination_id"]) if target is None: raise ReportingSubscriptionValidationError( - f"{where}.destination.destination_id {destination['destination_id']!r} is not in " + f"{where}.destination_id {destination['destination_id']!r} is not in " "reporting_destinations after this request" ) template = destination.get("path_template") @@ -187,36 +228,47 @@ def _validate_one( unknown = {name for name in _PLACEHOLDER.findall(template)} - _ALLOWED_PLACEHOLDERS if unknown: raise ReportingSubscriptionValidationError( - f"{where}.destination.path_template uses unknown placeholders {sorted(unknown)}" + f"{where}.path_template uses unknown placeholders {sorted(unknown)}" ) if "{" in _PLACEHOLDER.sub("", template) or "}" in _PLACEHOLDER.sub("", template): raise ReportingSubscriptionValidationError( - f"{where}.destination.path_template has braces outside a placeholder" + f"{where}.path_template has braces outside a placeholder" ) - if _multi_account(configuration): + overrides = configuration.get("destination_overrides") or () + if any(destination is item for item in overrides): + # An override routes only the accounts it names. + routes_many = len(destination.get("account_ids") or ()) > 1 or bool( + destination.get("operator_domains") + ) + else: + routes_many = _multi_account(configuration) + if routes_many: pattern = (target.get("configuration") or target).get("pattern") if template is None and pattern == "file_transfer": raise ReportingSubscriptionValidationError( - f"{where}.destination.path_template is required for a multi-account " - "file_transfer subscription" + f"{where}.path_template is required for a multi-account file_transfer subscription" ) if template is not None and "{account_id}" not in template: raise ReportingSubscriptionValidationError( - f"{where}.destination.path_template must contain {{account_id}} when the " + f"{where}.path_template must contain {{account_id}} when the " "scope can cover more than one account" ) def _validate_set(active: Sequence[dict[str, Any]], destinations: Mapping[str, Any]) -> None: - by_destination: dict[str, list[dict[str, Any]]] = {} + by_destination: dict[str, list[Mapping[str, Any]]] = {} for configuration in active: - destination = configuration.get("destination") - if destination is not None: - by_destination.setdefault(destination["destination_id"], []).append(configuration) + routes = [ + configuration.get("destination"), + *(configuration.get("destination_overrides") or ()), + ] + for route in routes: + if route is not None: + by_destination.setdefault(route["destination_id"], []).append(route) for destination_id, sharing in by_destination.items(): if len(sharing) < 2: continue - templates = [(item.get("destination") or {}).get("path_template") for item in sharing] + templates = [item.get("path_template") for item in sharing] if any(template is None or "{account_id}" not in template for template in templates): raise ReportingSubscriptionValidationError( f"active subscriptions sharing destination {destination_id!r} must each have a " @@ -229,13 +281,17 @@ def _validate_set(active: Sequence[dict[str, Any]], destinations: Mapping[str, A ) for first_index, first in enumerate(active): for second in active[first_index + 1 :]: - if first["selector"]["feed_purpose"] != second["selector"]["feed_purpose"]: + # adcp#8619: pacing and analytics allow several subscriptions per + # account and purpose; billing stays one per account per level. + if first["selector"]["feed_purpose"] != "billing": + continue + if second["selector"]["feed_purpose"] != "billing": continue if _same_level_overlap(first, second): raise ReportingSubscriptionValidationError( - f"active subscriptions {first['subscription_id']!r} and " - f"{second['subscription_id']!r} would both apply to one account and " - "feed_purpose at the same scope level" + f"active billing subscriptions {first['subscription_id']!r} and " + f"{second['subscription_id']!r} would both apply to one account at the " + "same scope level" ) diff --git a/src/adcp/reporting/subscriptions/resolution.py b/src/adcp/reporting/subscriptions/resolution.py index 4069702e3..3b294d0dc 100644 --- a/src/adcp/reporting/subscriptions/resolution.py +++ b/src/adcp/reporting/subscriptions/resolution.py @@ -80,6 +80,10 @@ def _wire_instant(value: Any, name: str) -> datetime | None: ) from error +def _optional_set(values: Any) -> frozenset[str] | None: + return None if values is None else frozenset(_text(item) for item in values) + + def _byte_key(offering_id: str) -> bytes: return offering_id.encode("utf-8") @@ -371,6 +375,24 @@ class SubscriptionSelector: alignment: str | None = None period_timezone: str | None = None destination_id: str | None = None + #: ``destination_overrides[]`` (adcp#8619): ``(account_ids, operator_domains, + #: destination_id)`` routes, in order. + destination_overrides: tuple[tuple[frozenset[str] | None, frozenset[str] | None, str], ...] = () + + def destination_for(self, account_id: str, operator: str | None = None) -> str | None: + """The ``destination_id`` an account's data goes to (adcp#8619). + + The ``account_ids`` override naming the account, else the + ``operator_domains`` override naming its operator, else ``destination``. + """ + for account_ids, _domains, destination_id in self.destination_overrides: + if account_ids is not None and account_id in account_ids: + return destination_id + if operator is not None: + for _ids, domains, destination_id in self.destination_overrides: + if domains is not None and operator in domains: + return destination_id + return self.destination_id @classmethod def from_subscription(cls, subscription: Any) -> SubscriptionSelector: @@ -409,6 +431,14 @@ def from_subscription(cls, subscription: Any) -> SubscriptionSelector: destination_id=( None if destination is None else _text(_value(destination, "destination_id")) ), + destination_overrides=tuple( + ( + _optional_set(_value(item, "account_ids")), + _optional_set(_value(item, "operator_domains")), + _text(_value(item, "destination_id")), + ) + for item in _value(subscription, "destination_overrides") or () + ), ) diff --git a/src/adcp/server/principal.py b/src/adcp/server/principal.py index 2de9aac12..e52579b73 100644 --- a/src/adcp/server/principal.py +++ b/src/adcp/server/principal.py @@ -109,6 +109,11 @@ class PrincipalChange: [PrincipalIdentity, AgentNotificationConfig], Awaitable[AgentNotificationConfigState] ] DeliveryConfigIdsHook = Callable[[PrincipalIdentity], Awaitable[Collection[str]]] +#: The full ``accounts_with_exceptions``, ``accounts_unauthorized`` or +#: ``prior_versions`` list of one subscription, for ``reporting_subscription_page``. +SubscriptionListHook = Callable[ + [PrincipalIdentity, str, str], Awaitable["Sequence[Mapping[str, Any]] | None"] +] SubscriptionVersionsHook = Callable[ [PrincipalIdentity, Sequence[ReportingSubscriptionState]], Awaitable[None] ] @@ -134,6 +139,9 @@ class ReportingSubscriptionOptions: #: (an engine built with ``end_account_configs``) sets this; otherwise the #: flag is refused with ``UNSUPPORTED_FEATURE``. replace_account_configs: bool = False + #: The full list behind an inline readback list (adcp#8619); without it, + #: ``reporting_subscription_page`` pages the inline list. + subscription_list: SubscriptionListHook | None = None @dataclass @@ -572,7 +580,92 @@ async def _get_principal_state( "context": request.context, } ) - return self._current_response(record, context=request.context) + response = self._current_response(record, context=request.context) + page = getattr(request, "reporting_subscription_page", None) + if page is None: + return response + return await self._subscription_page(identity, record, response, page, request.context) + + async def _subscription_page( + self, + identity: PrincipalIdentity, + record: PrincipalRecord, + response: GetPrincipalResponse, + page: Any, + context: object | None, + ) -> GetPrincipalResponse: + """One page of a subscription's readback list (adcp#8619).""" + request = page.model_dump(mode="json", exclude_none=True) + subscription_id = request["subscription_id"] + list_name = request["list"] + assert record.configuration is not None + state = next( + ( + item + for item in record.configuration.reporting_subscriptions or () + if item.subscription_id == subscription_id + ), + None, + ) + if state is None: + return GetPrincipalResponse.model_validate( + { + "status": "failed", + "result": { + "kind": "failed", + "errors": [ + { + "code": "INVALID_REQUEST", + "message": f"unknown reporting subscription {subscription_id!r}", + "field": "reporting_subscription_page.subscription_id", + } + ], + }, + "context": context, + } + ) + hook = self._subscriptions.subscription_list if self._subscriptions else None + items = None if hook is None else await hook(identity, subscription_id, list_name) + if items is None: + dumped = state.model_dump(mode="json", exclude_none=True) + items = dumped.get(list_name) or [] + pagination = request.get("pagination") or {} + limit = int(pagination.get("max_results") or 50) + cursor = pagination.get("cursor") + try: + offset = int(cursor) if cursor else 0 + except ValueError: + offset = -1 + if offset < 0 or offset > len(items): + return GetPrincipalResponse.model_validate( + { + "status": "failed", + "result": { + "kind": "failed", + "errors": [ + { + "code": "INVALID_REQUEST", + "message": "reporting_subscription_page.pagination.cursor is " + "not a cursor this seller issued", + "field": "reporting_subscription_page.pagination.cursor", + } + ], + }, + "context": context, + } + ) + following = offset + limit + page_payload: dict[str, Any] = { + "subscription_id": subscription_id, + "list": list_name, + list_name: [dict(item) for item in items[offset:following]], + "pagination": {"has_more": following < len(items), "total_count": len(items)}, + } + if following < len(items): + page_payload["pagination"]["cursor"] = str(following) + envelope = response.model_dump(mode="json", exclude_none=True) + envelope["reporting_subscription_page"] = page_payload + return GetPrincipalResponse.model_validate(envelope) async def sync_principal( self, diff --git a/tests/test_account_outcomes.py b/tests/test_account_outcomes.py index c9cdb3ef1..abbacd23e 100644 --- a/tests/test_account_outcomes.py +++ b/tests/test_account_outcomes.py @@ -93,6 +93,63 @@ async def test_requests_route_to_request_account_and_resolve_ready() -> None: assert client.adapter.request_account.call_count == 1 +@pytest.mark.asyncio +async def test_a_matched_account_carries_its_own_money_terms() -> None: + # adcp#8620: a matched or created account carries the account's money + # terms, which can differ from the terms requested. + client = _requests_client( + _record( + "resolved", + account_id="acc-1", + resolution="matched", + account={ + "account_id": "acc-1", + "name": "Acme", + "status": "active", + "billing": "agent", + "payment_terms": "net_45", + }, + ) + ) + outcome = await client.accounts.ensure(KEY, billing="operator") + assert isinstance(outcome, AccountReady) + assert outcome.resolution == "matched" + assert (outcome.billing, outcome.payment_terms) == ("agent", "net_45") + assert outcome.billing_entity is None + assert outcome.effective_at is None and outcome.not_carried_over == () + + +def test_an_updated_money_change_says_when_it_takes_effect() -> None: + from adcp.accounts import _request_outcome + + async def choose(request_id: str, account_id: str): # pragma: no cover - unused + raise AssertionError + + updated = _request_outcome( + _record( + "resolved", + account_id="acc-1", + resolution="updated", + effective_at="2026-12-01T00:00:00Z", + account={"account_id": "acc-1", "name": "Acme", "status": "active"}, + ), + choose=choose, + ) + assert isinstance(updated, AccountReady) + assert (updated.resolution, updated.effective_at) == ("updated", "2026-12-01T00:00:00Z") + separate = _request_outcome( + _record( + "resolved", + account_id="acc-2", + resolution="created", + not_carried_over=["notification_configs", "reporting_delivery_configs"], + account={"account_id": "acc-2", "name": "Acme EU", "status": "active"}, + ), + choose=choose, + ) + assert separate.not_carried_over == ("notification_configs", "reporting_delivery_configs") + + @pytest.mark.asyncio async def test_pending_is_a_value_and_a_repeat_ensure_never_opens_a_duplicate() -> None: client = _requests_client(_record("in_review", pending_action=SIGN_UP)) diff --git a/tests/test_principal_reporting_subscriptions.py b/tests/test_principal_reporting_subscriptions.py index d39485aaf..d7e16cead 100644 --- a/tests/test_principal_reporting_subscriptions.py +++ b/tests/test_principal_reporting_subscriptions.py @@ -212,10 +212,22 @@ async def test_new_version_records_prior_versions() -> None: async def test_reused_version_with_new_semantics_is_rejected() -> None: principals, _clock = service() await sync(principals, {"reporting_subscriptions": [subscription()]}) - changed = subscription(exclude_account_ids=["acct-9"]) + changed = subscription( + selector={ + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "dimensions": ["package", "creative"], + } + ) assert error_code(await sync(principals, {"reporting_subscriptions": [changed]})) == ( "INVALID_REQUEST" ) + # Changing the scope kind still needs a new version. + narrowed = subscription(scope={"account_ids": ["acct-9"]}) + assert error_code(await sync(principals, {"reporting_subscriptions": [narrowed]})) == ( + "INVALID_REQUEST" + ) older = subscription(subscription_version=1) await sync(principals, {"reporting_subscriptions": [subscription(subscription_version=2)]}) assert error_code(await sync(principals, {"reporting_subscriptions": [older]})) == ( @@ -223,6 +235,70 @@ async def test_reused_version_with_new_semantics_is_rejected() -> None: ) +@pytest.mark.asyncio +async def test_membership_edits_apply_in_place() -> None: + # adcp#8619: adding or removing account_ids and changing exclude_account_ids + # keep subscription_version and accepted_at. + principals, clock = service() + named = subscription(scope={"account_ids": ["acct-1"]}) + await sync(principals, {"reporting_subscriptions": [named]}) + clock.advance(hours=5) + widened = subscription(scope={"account_ids": ["acct-1", "acct-2"]}) + state = states(await sync(principals, {"reporting_subscriptions": [widened]}))[ + "analytics-daily" + ] + assert state["subscription_version"] == 1 + assert state["accepted_at"] == "2026-10-01T12:00:00Z" + assert state["configuration"]["scope"]["account_ids"] == ["acct-1", "acct-2"] + assert "prior_versions" not in state + excluded = subscription("all", exclude_account_ids=["acct-9"]) + await sync(principals, {"reporting_subscriptions": [widened, excluded]}) + clock.advance(hours=1) + reincluded = subscription("all") + state = states(await sync(principals, {"reporting_subscriptions": [widened, reincluded]}))[ + "all" + ] + assert state["subscription_version"] == 1 + assert "exclude_account_ids" not in state["configuration"] + + +@pytest.mark.asyncio +async def test_destination_overrides_name_existing_destinations() -> None: + principals, _clock = service() + routed = subscription( + destination={"destination_id": "archive", "path_template": "a/{account_id}"}, + destination_overrides=[ + { + "account_ids": ["acct-1"], + "destination_id": "warehouse", + "path_template": "w/{period_start}", + } + ], + ) + applied = await sync( + principals, + {"reporting_destinations": [DESTINATION, WAREHOUSE], "reporting_subscriptions": [routed]}, + ) + assert applied.result.kind == "applied", applied + missing = await sync( + principals, + {"reporting_destinations": [DESTINATION], "reporting_subscriptions": [routed]}, + ) + assert error_code(missing) == "INVALID_REQUEST" + shared = subscription( + destination={"destination_id": "archive", "path_template": "a/{account_id}"}, + destination_overrides=[ + {"operator_domains": ["agency.example"], "destination_id": "warehouse"} + ], + ) + unkeyed = await sync( + principals, + {"reporting_destinations": [DESTINATION, WAREHOUSE], "reporting_subscriptions": [shared]}, + ) + # An operator override can route many accounts: it needs {account_id}. + assert error_code(unkeyed) == "INVALID_REQUEST" + + @pytest.mark.asyncio async def test_active_changes_in_place() -> None: principals, clock = service() @@ -437,11 +513,39 @@ async def test_set_level_rules() -> None: principals, {"reporting_subscriptions": [subscription(), subscription()]} ) assert error_code(duplicate) == "INVALID_REQUEST" + # adcp#8619: pacing and analytics allow several subscriptions per account + # and purpose (an hourly and a daily feed, or two breakdowns). two_all = await sync( principals, - {"reporting_subscriptions": [subscription("a"), subscription("b")]}, + {"reporting_subscriptions": [subscription("hourly"), subscription("daily")]}, + ) + assert two_all.result.kind == "applied" + + def billing(subscription_id: str, **overrides) -> dict: + return subscription( + subscription_id, + selector={ + "feed_purpose": "billing", + "period_duration": "P1D", + "required_finality": "official", + "dimensions": ["package"], + }, + reconciliation_mode="consumer_receipt", + destination={ + "destination_id": "archive", + "path_template": f"{subscription_id}/{{account_id}}", + }, + **overrides, + ) + + two_billing = await sync( + principals, + { + "reporting_destinations": [DESTINATION], + "reporting_subscriptions": [billing("a"), billing("b")], + }, ) - assert error_code(two_all) == "INVALID_REQUEST" + assert error_code(two_billing) == "INVALID_REQUEST" disjoint_operators = await sync( principals, { @@ -455,10 +559,11 @@ async def test_set_level_rules() -> None: shared_accounts = await sync( principals, { + "reporting_destinations": [DESTINATION], "reporting_subscriptions": [ - subscription("a", scope={"account_ids": ["x", "y"]}), - subscription("b", scope={"account_ids": ["y"]}), - ] + billing("a", scope={"account_ids": ["x", "y"]}), + billing("b", scope={"account_ids": ["y"]}), + ], }, ) assert error_code(shared_accounts) == "INVALID_REQUEST" diff --git a/tests/test_reporting_subscription_beta3_engine.py b/tests/test_reporting_subscription_beta3_engine.py new file mode 100644 index 000000000..6eb5e215e --- /dev/null +++ b/tests/test_reporting_subscription_beta3_engine.py @@ -0,0 +1,285 @@ +"""Seller-engine rules settled for AdCP 3.3.0-beta.3 (adcp#8619). + +Admission under ``max_accounts``, several pacing or analytics subscriptions per +account and purpose, membership edits in place, and ``destination_overrides``. +The buyer computes the same owed periods from the wire, so each test checks +the seller against :func:`owed_subscription_periods` where it can. +""" + +from __future__ import annotations + +import pytest + +from adcp.reporting.subscriptions import ( + ReportingSubscriptionSupport, + SubscriptionBuyPackage, + SubscriptionDestination, + SubscriptionVersionWindow, + owed_subscription_periods, +) +from tests.test_reporting_subscription_engine import ( + ALL_IDS, + World, + at, + buy, + offering, + subscription, +) + + +def _packaged(media_buy_id: str, start: str, end: str): + return buy( + media_buy_id, + start, + end, + offering_ids=(), + packages=(SubscriptionBuyPackage("pkg-1", ALL_IDS),), + ) + + +def _starts(periods) -> list[str]: + return [item.period_start for item in periods] + + +@pytest.mark.asyncio +async def test_admission_orders_by_authorized_at_then_account_id() -> None: + world = World(support=ReportingSubscriptionSupport(max_accounts=1)) + # acct-a sorts first by id, but acct-b was authorized first. + world.grant("acct-b", "2026-09-01T00:00:00Z") + world.grant("acct-a", "2026-09-15T00:00:00Z") + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + admitted = await world.coverage("acct-b", "analytics") + waiting = await world.coverage("acct-a", "analytics") + assert admitted.status.value == "active" + assert waiting.status.value == "excluded" + assert [item.reason.value for item in waiting.excluded] == ["capacity"] + # Sticky: a candidate sorting first never displaces the admitted account. + world.grant("acct-0", "2026-08-01T00:00:00Z") + await world.reconcile("2026-10-02T00:00:00Z") + assert (await world.coverage("acct-b", "analytics")).status.value == "active" + assert (await world.coverage("acct-0", "analytics")).status.value == "excluded" + + +@pytest.mark.asyncio +async def test_two_analytics_subscriptions_resolving_one_offering_owe_it_once() -> None: + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = [_packaged("mb-1", "2026-10-01T04:00:00Z", "2026-10-04T04:00:00Z")] + hourly = subscription( + "c-hourly", + selector={ + "feed_purpose": "analytics", + "period_duration": "PT1H", + "required_finality": "official", + "dimensions": ["package"], + }, + ) + await world.sync(subscription("a-daily"), subscription("b-daily"), hourly) + await world.reconcile("2026-10-01T00:00:00Z") + # The lowest subscription_id owes the shared offering; the other records a + # package-level overridden exception naming it. + second = await world.coverage("acct-1", "b-daily") + (entry,) = second.excluded + wire = entry.model_dump(mode="json", exclude_none=True) + assert wire["reason"] == "overridden" + assert wire["media_buy_id"] == "mb-1" + assert wire["package_ids"] == ["pkg-1"] + assert wire["overridden_by"] == {"subscription_id": "a-daily"} + assert second.status.value == "active" # a package-level entry, not account-level + now = "2026-10-05T00:00:00Z" + assert _starts(await world.seller_owes("a-daily", "acct-1", now)) == [ + "2026-10-01T04:00:00Z", + "2026-10-02T04:00:00Z", + "2026-10-03T04:00:00Z", + ] + assert await world.seller_owes("b-daily", "acct-1", now) == [] + # A different offering of the same purpose is owed alongside. + assert await world.coverage("acct-1", "c-hourly") + assert len(await world.seller_owes("c-hourly", "acct-1", now)) == 72 + + +@pytest.mark.asyncio +async def test_removing_an_account_in_place_keeps_earlier_periods_owed() -> None: + world = World() + for account_id in ("acct-1", "acct-2"): + world.grant(account_id, "2026-09-01T00:00:00Z") + world.host.buys[account_id] = [ + buy(f"mb-{account_id}", "2026-10-01T04:00:00Z", "2026-10-06T04:00:00Z") + ] + both = subscription(scope={"account_ids": ["acct-1", "acct-2"]}) + await world.sync(both) + await world.reconcile("2026-10-01T00:00:00Z") + world.clock.set("2026-10-02T12:00:00Z") + # Same version: the membership edit applies in place. + await world.sync(subscription(scope={"account_ids": ["acct-1"]})) + await world.reconcile("2026-10-02T12:00:00Z") + state = await world.state("analytics") + assert state.subscription_version == 1 and state.prior_versions is None + removed = await world.coverage("acct-2", "analytics") + assert removed is not None and removed.status.value == "excluded" + (entry,) = removed.excluded + wire = entry.model_dump(mode="json", exclude_none=True) + # From E: the Oct 2 New York day started before the edit and stays owed. + assert wire == {"reason": "excluded_by_buyer", "from": "2026-10-03T04:00:00Z"} + owed = await world.seller_owes("analytics", "acct-2", "2026-10-07T00:00:00Z") + assert _starts(owed) == ["2026-10-01T04:00:00Z", "2026-10-02T04:00:00Z"] + # The buyer computes the same from the readback and the coverage entry. + assert _starts( + await world.buyer_expects("analytics", "acct-2", "2026-10-07T00:00:00Z", removed) + ) == _starts(owed) + + +@pytest.mark.asyncio +async def test_adding_an_account_in_place_owes_only_from_when_it_joined() -> None: + world = World() + for account_id in ("acct-1", "acct-2"): + world.grant(account_id, "2026-09-01T00:00:00Z") + world.host.buys[account_id] = [ + buy(f"mb-{account_id}", "2026-10-01T04:00:00Z", "2026-10-06T04:00:00Z") + ] + await world.sync(subscription(scope={"account_ids": ["acct-1"]})) + await world.reconcile("2026-10-01T00:00:00Z") + world.clock.set("2026-10-03T12:00:00Z") + await world.sync(subscription(scope={"account_ids": ["acct-1", "acct-2"]})) + await world.reconcile("2026-10-03T12:00:00Z") + added = await world.coverage("acct-2", "analytics") + assert added.status.value == "active" + (entry,) = added.excluded + wire = entry.model_dump(mode="json", exclude_none=True) + # One closed entry from the first boundary at or after accepted_at to the + # first boundary at or after the edit. + assert wire == { + "reason": "excluded_by_buyer", + "from": "2026-10-01T04:00:00Z", + "until": "2026-10-04T04:00:00Z", + } + owed = await world.seller_owes("analytics", "acct-2", "2026-10-07T00:00:00Z") + assert _starts(owed) == ["2026-10-04T04:00:00Z", "2026-10-05T04:00:00Z"] + assert _starts( + await world.buyer_expects("analytics", "acct-2", "2026-10-07T00:00:00Z", added) + ) == _starts(owed) + + +def test_an_override_routes_one_account_to_another_destination_method() -> None: + to_s3 = offering("pkg-s3", ("package",), method_pattern="file_transfer", method_transport="s3") + to_gcs = offering( + "pkg-gcs", ("package",), method_pattern="file_transfer", method_transport="gcs" + ) + configuration = subscription( + destination={"destination_id": "lake", "path_template": "l/{account_id}"}, + destination_overrides=[{"account_ids": ["acct-1"], "destination_id": "warehouse"}], + ) + windows = [SubscriptionVersionWindow(1, at("2026-10-01T00:00:00Z"), None, configuration)] + destinations = { + "lake": SubscriptionDestination("file_transfer", "s3"), + "warehouse": SubscriptionDestination("file_transfer", "gcs"), + } + kwargs = dict( + subscription=windows, + buys=[ + buy( + "mb-1", + "2026-10-01T04:00:00Z", + "2026-10-02T04:00:00Z", + offering_ids=("pkg-s3", "pkg-gcs"), + ) + ], + offerings=[to_s3, to_gcs], + account_timezone="America/New_York", + now=at("2026-10-03T00:00:00Z"), + destinations=destinations, + ) + routed = owed_subscription_periods(account_id="acct-1", **kwargs) + default = owed_subscription_periods(account_id="acct-2", **kwargs) + assert {item.offering_id for item in routed} == {"pkg-gcs"} + assert {item.offering_id for item in default} == {"pkg-s3"} + + +def test_the_engine_routes_a_version_to_the_override_for_managed_binding() -> None: + from adcp.reporting.subscriptions.engine import _Account, _routed_version + + configuration = subscription( + destination={"destination_id": "lake", "path_template": "l/{account_id}"}, + destination_overrides=[ + { + "operator_domains": ["agency.example"], + "destination_id": "warehouse", + "path_template": "w/{account_id}", + } + ], + ) + agency = _Account(timezone="UTC", operator="agency.example") + direct = _Account(timezone="UTC", operator="brand.example") + assert _routed_version(configuration, "acct-1", agency)["destination"] == { + "destination_id": "warehouse", + "path_template": "w/{account_id}", + } + assert _routed_version(configuration, "acct-1", direct) is configuration + + +async def _page(world: World, subscription_id: str, list_name: str, **pagination): + from tests.test_reporting_subscription_engine import IDENTITY + + request = { + "reporting_subscription_page": {"subscription_id": subscription_id, "list": list_name} + } + if pagination: + request["reporting_subscription_page"]["pagination"] = pagination + response = await world.principals.get_principal(IDENTITY, request) + return response.model_dump(mode="json", exclude_none=True) + + +@pytest.mark.asyncio +async def test_prior_versions_list_sixteen_inline_and_page_the_rest() -> None: + world = World() + for version in range(1, 19): + world.clock.set(f"2026-10-{version:02d}T00:00:00Z") + await world.sync(subscription(subscription_version=version)) + state = await world.state("analytics") + assert [item.subscription_version for item in state.prior_versions] == list(range(17, 1, -1)) + assert state.prior_versions_count == 17 + first = await _page(world, "analytics", "prior_versions", max_results=10) + page = first["reporting_subscription_page"] + assert [item["subscription_version"] for item in page["prior_versions"]] == list( + range(17, 7, -1) + ) + assert page["pagination"] == {"has_more": True, "cursor": "10", "total_count": 17} + rest = await _page(world, "analytics", "prior_versions", max_results=10, cursor="10") + page = rest["reporting_subscription_page"] + assert [item["subscription_version"] for item in page["prior_versions"]] == list( + range(7, 0, -1) + ) + assert page["pagination"] == {"has_more": False, "total_count": 17} + # The usual result rides along. + assert rest["result"]["kind"] == "current" + + +@pytest.mark.asyncio +async def test_an_account_list_pages_in_account_id_order_with_its_count() -> None: + world = World() + for account_id in ("acct-c", "acct-a", "acct-b"): + world.grant(account_id, "2026-09-01T00:00:00Z") + world.host.buys[account_id] = [ + buy(f"mb-{account_id}", "2026-10-01T04:00:00Z", "2026-10-03T04:00:00Z") + ] + await world.sync(subscription()) + await world.reconcile("2026-10-01T00:00:00Z") + for account_id in ("acct-c", "acct-a", "acct-b"): + world.revoke(account_id, "2026-10-02T00:00:00Z") + await world.reconcile("2026-10-02T00:00:00Z") + state = await world.state("analytics") + assert state.accounts_unauthorized_count == 3 + first = await _page(world, "analytics", "accounts_unauthorized", max_results=2) + page = first["reporting_subscription_page"] + assert [item["account_id"] for item in page["accounts_unauthorized"]] == ["acct-a", "acct-b"] + assert page["pagination"] == {"has_more": True, "cursor": "2", "total_count": 3} + + +@pytest.mark.asyncio +async def test_paging_an_unknown_subscription_is_invalid() -> None: + world = World() + await world.sync(subscription()) + response = await _page(world, "missing", "prior_versions") + assert response["result"]["kind"] == "failed" + assert response["result"]["errors"][0]["code"] == "INVALID_REQUEST" From 1a34cfa20bdd6983ee3a4b6a6a517ad69f0c5e37 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sat, 31 Oct 2026 13:59:26 +0000 Subject: [PATCH 2/2] feat(reporting): route destinations per account as the TypeScript SDK does reporting_subscription_destination_for_account, the TypeScript reportingSubscriptionDestinationForAccount, picks the account_ids override naming the account, else the operator_domains override naming its operator, else destination. An override without a path_template keeps the subscription's. It is exported from adcp.reporting.subscriptions and adcp.reporting.advanced, and the managed binding routes each account's version through it. Registration refuses an override that names both account_ids and operator_domains, or neither, which the generated model does not enforce. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/advanced.py | 2 + src/adcp/reporting/subscriptions/__init__.py | 2 + src/adcp/reporting/subscriptions/engine.py | 28 +++++------- src/adcp/reporting/subscriptions/registry.py | 45 +++++++++++++++++++ .../test_principal_reporting_subscriptions.py | 18 ++++++++ ...est_reporting_subscription_beta3_engine.py | 27 +++++++++++ 6 files changed, 104 insertions(+), 18 deletions(-) diff --git a/src/adcp/reporting/advanced.py b/src/adcp/reporting/advanced.py index 1deb2ca36..fa2cb1f48 100644 --- a/src/adcp/reporting/advanced.py +++ b/src/adcp/reporting/advanced.py @@ -84,6 +84,7 @@ apply_reporting_subscriptions, refresh_destination_states, render_path_template, + reporting_subscription_destination_for_account, ) from adcp.reporting.subscriptions.resolution import ( STANDARD_REPORTING_DIMENSIONS, @@ -160,6 +161,7 @@ "apply_reporting_subscriptions", "refresh_destination_states", "render_path_template", + "reporting_subscription_destination_for_account", # Service planner hooks "ManagedGenerationBinder", "PlanningErrorHandler", diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index db577c844..a79aaa0c8 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -95,6 +95,7 @@ apply_reporting_subscriptions, refresh_destination_states, render_path_template, + reporting_subscription_destination_for_account, ) from adcp.reporting.subscriptions.resolution import ( STANDARD_REPORTING_DIMENSIONS, @@ -160,6 +161,7 @@ "apply_reporting_subscriptions", "refresh_destination_states", "render_path_template", + "reporting_subscription_destination_for_account", "REPORTING_SUBSCRIPTIONS_FEATURE", "REPORTING_SUBSCRIPTIONS_SECTION", "STANDARD_REPORTING_DIMENSIONS", diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index a45d89b05..e7e584742 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -71,6 +71,9 @@ subscription_covers_account, subscription_version_windows, ) +from adcp.reporting.subscriptions.registry import ( + reporting_subscription_destination_for_account, +) from adcp.reporting.subscriptions.resolution import ( SubscriptionDestination, SubscriptionOffering, @@ -2449,25 +2452,14 @@ def _routed_version( The managed binding then resolves and checks the override's ``destination_id`` and ``path_template`` (adcp#8619). """ - for override in configuration.get("destination_overrides") or (): - account_ids = override.get("account_ids") - if account_ids is not None and account_id in account_ids: - break - else: - override = next( - ( - item - for item in configuration.get("destination_overrides") or () - if item.get("operator_domains") and account.operator in item["operator_domains"] - ), - None, - ) - if override is None: + if not configuration.get("destination_overrides"): + return configuration + routed = reporting_subscription_destination_for_account( + configuration, account_id=account_id, operator=account.operator + ) + if routed is None or routed == configuration.get("destination"): return configuration - destination = {"destination_id": override["destination_id"]} - if override.get("path_template") is not None: - destination["path_template"] = override["path_template"] - return {**configuration, "destination": destination} + return {**configuration, "destination": routed} def _member_ids(configuration: Mapping[str, Any]) -> set[str] | None: diff --git a/src/adcp/reporting/subscriptions/registry.py b/src/adcp/reporting/subscriptions/registry.py index 27c18d882..52cdc2c1d 100644 --- a/src/adcp/reporting/subscriptions/registry.py +++ b/src/adcp/reporting/subscriptions/registry.py @@ -196,6 +196,13 @@ def _validate_one( f"{where}: consumer_receipt requires a destination" ) for override_index, override in enumerate(configuration.get("destination_overrides") or ()): + # The schema's anyOf/not: exactly one of account_ids and operator_domains, + # which the generated model does not enforce. + if ("account_ids" in override) == ("operator_domains" in override): + raise ReportingSubscriptionValidationError( + f"{where}.destination_overrides[{override_index}] must name exactly one of " + "account_ids or operator_domains" + ) _validate_destination( override, f"{where}.destination_overrides[{override_index}]", @@ -211,6 +218,38 @@ def _validate_one( _validate_destination(destination, f"{where}.destination", configuration, destinations) +def reporting_subscription_destination_for_account( + subscription: Mapping[str, Any] | Any, *, account_id: str, operator: str | None = None +) -> dict[str, Any] | None: + """The destination one account's data goes to under a subscription (adcp#8619). + + The ``account_ids`` override naming the account, else the + ``operator_domains`` override naming its operator, else ``destination``. + An override without its own ``path_template`` keeps the subscription's. + ``None`` when the subscription names no destination. The TypeScript SDK's + ``reportingSubscriptionDestinationForAccount``. + """ + configuration = _dump(subscription) + destination = configuration.get("destination") + if destination is None: + return None + overrides = configuration.get("destination_overrides") or () + chosen = next( + (item for item in overrides if account_id in (item.get("account_ids") or ())), None + ) + if chosen is None and operator is not None: + chosen = next( + (item for item in overrides if operator in (item.get("operator_domains") or ())), None + ) + if chosen is None: + return dict(destination) + template = chosen.get("path_template", destination.get("path_template")) + routed: dict[str, Any] = {"destination_id": chosen["destination_id"]} + if template is not None: + routed["path_template"] = template + return routed + + def _validate_destination( destination: Mapping[str, Any], where: str, @@ -224,6 +263,11 @@ def _validate_destination( "reporting_destinations after this request" ) template = destination.get("path_template") + if template is None and any( + destination is item for item in configuration.get("destination_overrides") or () + ): + # An override without its own path_template keeps the subscription's. + template = (configuration.get("destination") or {}).get("path_template") if template is not None: unknown = {name for name in _PLACEHOLDER.findall(template)} - _ALLOWED_PLACEHOLDERS if unknown: @@ -568,6 +612,7 @@ def refresh_destination_states( __all__ = [ + "reporting_subscription_destination_for_account", "ReportingSubscriptionValidationError", "apply_reporting_subscriptions", "refresh_destination_states", diff --git a/tests/test_principal_reporting_subscriptions.py b/tests/test_principal_reporting_subscriptions.py index d7e16cead..421ded31c 100644 --- a/tests/test_principal_reporting_subscriptions.py +++ b/tests/test_principal_reporting_subscriptions.py @@ -297,6 +297,24 @@ async def test_destination_overrides_name_existing_destinations() -> None: ) # An operator override can route many accounts: it needs {account_id}. assert error_code(unkeyed) == "INVALID_REQUEST" + both = subscription( + destination={"destination_id": "archive", "path_template": "a/{account_id}"}, + destination_overrides=[ + { + "account_ids": ["acct-1"], + "operator_domains": ["agency.example"], + "destination_id": "warehouse", + "path_template": "w/{account_id}", + } + ], + ) + refused = await sync( + principals, + {"reporting_destinations": [DESTINATION, WAREHOUSE], "reporting_subscriptions": [both]}, + ) + # Exactly one of account_ids or operator_domains (the generated model + # does not enforce the schema's anyOf/not). + assert error_code(refused) == "INVALID_REQUEST" @pytest.mark.asyncio diff --git a/tests/test_reporting_subscription_beta3_engine.py b/tests/test_reporting_subscription_beta3_engine.py index 6eb5e215e..8d5f82104 100644 --- a/tests/test_reporting_subscription_beta3_engine.py +++ b/tests/test_reporting_subscription_beta3_engine.py @@ -283,3 +283,30 @@ async def test_paging_an_unknown_subscription_is_invalid() -> None: response = await _page(world, "missing", "prior_versions") assert response["result"]["kind"] == "failed" assert response["result"]["errors"][0]["code"] == "INVALID_REQUEST" + + +def test_the_destination_for_an_account_follows_the_override_order() -> None: + from adcp.reporting.subscriptions import reporting_subscription_destination_for_account + + configuration = subscription( + destination={"destination_id": "lake", "path_template": "l/{account_id}"}, + destination_overrides=[ + {"operator_domains": ["agency.example"], "destination_id": "warehouse"}, + {"account_ids": ["acct-1"], "destination_id": "vault", "path_template": "v/x"}, + ], + ) + + def route(account_id: str, operator: str | None = None): + return reporting_subscription_destination_for_account( + configuration, account_id=account_id, operator=operator + ) + + # account_ids wins over operator_domains, whatever the order listed. + assert route("acct-1", "agency.example") == {"destination_id": "vault", "path_template": "v/x"} + # An override without a path_template keeps the subscription's. + assert route("acct-2", "agency.example") == { + "destination_id": "warehouse", + "path_template": "l/{account_id}", + } + assert route("acct-2") == {"destination_id": "lake", "path_template": "l/{account_id}"} + assert reporting_subscription_destination_for_account(subscription(), account_id="a") is None