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/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/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..e7e584742 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 @@ -58,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, @@ -324,6 +340,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 +629,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 +723,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 +845,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 +1130,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 +1518,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 +1535,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 +1573,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 +1590,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 +1620,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 +1637,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 +1649,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 +1665,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 +2025,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 +2224,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 +2308,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 +2422,86 @@ 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). + """ + 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 + return {**configuration, "destination": routed} + + +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..52cdc2c1d 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,49 +195,124 @@ def _validate_one( raise ReportingSubscriptionValidationError( 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}]", + 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 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, + 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") + 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: 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 +325,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" ) @@ -512,6 +612,7 @@ def refresh_destination_states( __all__ = [ + "reporting_subscription_destination_for_account", "ReportingSubscriptionValidationError", "apply_reporting_subscriptions", "refresh_destination_states", 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..421ded31c 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,88 @@ 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" + 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 async def test_active_changes_in_place() -> None: principals, clock = service() @@ -437,11 +531,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 +577,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..8d5f82104 --- /dev/null +++ b/tests/test_reporting_subscription_beta3_engine.py @@ -0,0 +1,312 @@ +"""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" + + +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