From 6657c75a18c86420192b3ec0cc4aa97e7c697293 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 17:00:43 +0000 Subject: [PATCH 1/6] feat(reporting): explain coverage with a shared CoverageTrace Add explain_subscription_coverage, which classifies periods of one subscription on one account against the facts owed_subscription_periods reads. The verdict is the owed rule's own, and the reason is the first rule, in the spec's fixed order, that rejects the period, with the facts it read. Reason codes, verdicts and rules come from the spec's reason-codes.json in the packaged compliance tree, and explanations render from one template per reason, so buyer and seller print identical text for identical facts. The CoverageTrace JSON carries trace_version, side, subject, verdict, reason, explanation, steps and an inputs_digest over wire-visible facts local to the period. diff_coverage_traces names the first differing rule and whether the two sides read the same inputs. A seller-only block carries buy visibility and the obligation id, and for_buyer strips it. Seller surface: ReliableReportingService.explain_coverage and ReportingSubscriptionEngine.explain_coverage, including buy_not_visible for buys the caller cannot see. CLI: python -m adcp reporting explain, against the seller service named by --app, and python -m adcp reporting diff. The coverage input normalization moves into a shared context, so the owed rule and explain read one preparation. Every canonical vector is checked: each period near each buy is explained, and the verdict equals the vector's expected owed rows. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reporting-subscriptions-coverage.md | 129 +- docs/reporting-subscriptions-seller.md | 8 +- src/adcp/__main__.py | 5 + src/adcp/reporting/cli.py | 166 +++ src/adcp/reporting/service.py | 39 + src/adcp/reporting/subscriptions/__init__.py | 23 + src/adcp/reporting/subscriptions/coverage.py | 210 ++- src/adcp/reporting/subscriptions/engine.py | 104 ++ src/adcp/reporting/subscriptions/explain.py | 1378 ++++++++++++++++++ tests/test_reporting_subscription_explain.py | 443 ++++++ 10 files changed, 2423 insertions(+), 82 deletions(-) create mode 100644 src/adcp/reporting/cli.py create mode 100644 src/adcp/reporting/subscriptions/explain.py create mode 100644 tests/test_reporting_subscription_explain.py diff --git a/docs/reporting-subscriptions-coverage.md b/docs/reporting-subscriptions-coverage.md index 242dfe4e4..e2018fc35 100644 --- a/docs/reporting-subscriptions-coverage.md +++ b/docs/reporting-subscriptions-coverage.md @@ -15,43 +15,134 @@ A period of a resolved offering is owed for an account when all of these hold: The rules are applied in a fixed order: `scope`, `authorization`, `version`, `activity`, `offering`, `flight`, `confirmation`, `exception`, `gap`, `cutover`, `timezone`. +## Explaining a period + +`explain` answers "why is this period owed, or not?" with a `CoverageTrace`: the verdict, the reason code, a one-sentence explanation, every rule applied in order with the facts it read, and an `inputs_digest`. + +```python +# Seller: from your service and storage. +traces = await reporting.explain_coverage( + PrincipalRef("t1", "agency-x"), "acc_renault_fr", + subscription_id="analytics-daily", media_buy_id="mb_123", period="2026-11-03") +for trace in traces: + print(trace.verdict, trace.reason, trace.explanation) +``` + +``` +python -m adcp reporting explain --app myapp.reporting:service \ + --principal agency-x --account acc_renault_fr --period 2026-11-03 [--json] +``` + +- `period` is a `YYYY-MM-DD` date read in each offering's zone (every period that starts on that local date), an instant (the period containing it), or `{"start", "end"}`, which must be a period of the offering's schedule. The trace echoes the period with its `timezone`. +- Omit `subscription_id`, `media_buy_id` or `offering_id` to explain every match. The result is a list, sorted by period start, buy and offering. +- The verdict is the owed rule's own: a period is `owed` exactly when `owed_subscription_periods` owes it. The reason is the first rule, in the fixed order, that rejects the period. +- `--app` names your `ReliableReportingService` as `module:attribute`, the way ASGI servers name an app. + +### The trace + +```json +{ + "trace_version": 1, + "side": "seller", + "subject": { + "principal_id": "agency-x", "account_id": "acc_a", + "subscription_id": "analytics-daily", "subscription_version": 1, + "media_buy_id": "mb_123", "offering_id": "daily_pkg", + "period": {"start": "2026-11-03T05:00:00Z", "end": "2026-11-04T05:00:00Z", "timezone": "America/New_York"} + }, + "verdict": "not_owed", + "reason": "before_confirmation", + "explanation": "Not owed: the period ends 2026-11-04T05:00Z, before mb_123 was confirmed at 2026-11-04T10:00Z.", + "steps": [ + {"rule": "scope", "result": "pass", "facts": {"scope": "all_authorized_accounts", "operator_domains": null, "operator": null}}, + {"rule": "authorization", "result": "pass", "facts": {"authorized_at": null, "chain": "not_advertised"}}, + {"rule": "version", "result": "pass", "facts": {"subscription_version": 1, "accepted_at": "2026-10-20T12:00:00Z", "coverage_start": "2026-10-21T04:00:00Z"}}, + {"rule": "activity", "result": "pass", "facts": {"inactive_intervals": []}}, + {"rule": "offering", "result": "pass", "facts": {"match": "exact", "candidates": ["daily_pkg"], "resolved": ["daily_pkg"], "selected": "daily_pkg"}}, + {"rule": "flight", "result": "pass", "facts": {"start_time": "2026-11-03T05:00:00Z", "stop_time": "2026-11-30T05:00:00Z", "as_of": "2026-11-04T05:00:00Z"}}, + {"rule": "confirmation", "result": "fail", "facts": {"confirmed_at": "2026-11-04T10:00:00Z"}} + ], + "inputs_digest": "sha256:…", + "seller_only": {"buy_visible": true} +} +``` + +- Steps stop at the first failing rule. An owed period passes every rule; `timezone` is `not_applicable` unless the offering is `account_timezone` aligned. +- `seller_only` holds whether the caller can see the buy and, for an owed period, its `reporting_obligation_id`. It never leaves the seller: send `trace.for_buyer()`. +- `inputs_digest` hashes only wire-visible facts local to the period: the version in effect, the authorization as `list_accounts` and `accounts_unauthorized` show it, the zone in effect, the buy as `get_media_buys` returns it, the offering, the `excluded[]` entries covering the period start, and whether an authorization gap covers it. `now` is not part of it. + +### Comparing a buyer trace with a seller trace + +``` +python -m adcp reporting diff buyer.json seller.json +``` + +`diff_coverage_traces(left, right)` names the first rule whose result or facts differ: +- **Equal digests and different verdicts:** both sides read the same facts, so one SDK applied a rule differently. Report it. +- **Different digests:** one side read stale or different state. Re-read and compare again. + ## Reason codes -The frozen vocabulary (`COVERAGE_REASON_CODES`, from the spec's `reason-codes.json`): - -| Reason | Rule | -|---|---| -| `not_in_scope`, `excluded:excluded_by_buyer`, `buy_not_visible` (seller only) | scope | -| `not_authorized` | authorization | -| `before_coverage_start` | version | -| `subscription_inactive` | activity | -| `excluded:offering_unsupported` | offering | -| `outside_flight`, `after_stop` | flight | -| `before_confirmation` | confirmation | -| `excluded:capacity`, `excluded:overridden`, `excluded:action_required` | exception | -| `gap:unauthorized` | gap | -| `cutover_stretch`, `owed_under_other_version` | cutover | -| `period_not_ended`, `owed`, `owed:regrant_backfill`, `owed:withheld` | outcome | +The frozen vocabulary comes from the spec's `reason-codes.json`, packaged with the SDK and read by `coverage_reasons()`. `COVERAGE_REASON_CODES` mirrors it. Each code renders one sentence from `EXPLAIN_TEMPLATES`, so both sides print identical text for identical facts. + +| Reason | Verdict | Rule | Sentence | +|---|---|---|---| +| `not_in_scope` | not owed | scope | Not owed: version {v} of {subscription} does not select account {account}. | +| `excluded:excluded_by_buyer` | not owed | scope | Not owed: the buyer excluded account {account} from {subscription} from {from}. | +| `buy_not_visible` (seller only) | not owed | scope | Not owed: the caller cannot see {buy} through get_media_buys. | +| `not_authorized` | not owed | authorization | Not owed: the caller held no authority for {account} over the period {start} to {end}. | +| `before_coverage_start` | not owed | version | Not owed: the period starts {start}, before coverage starts at {coverage_start}. | +| `subscription_inactive` | not owed | activity | Not owed: {subscription} was inactive from {deactivated_at} when the period started at {start}. | +| `excluded:offering_unsupported` | not owed | offering | Not owed: {offering} does not report {buy} under the selector of version {v}. | +| `outside_flight` | not owed | flight | Not owed: the period ends {end}, at or before {buy} starts at {start_time}. | +| `after_stop` | not owed | flight | Not owed: the period starts {start}, at or after {buy} stopped at {stop_time}. | +| `before_confirmation` | not owed | confirmation | Not owed: the period ends {end}, before {buy} was confirmed at {confirmed_at}. | +| `excluded:capacity` | not owed | exception | Not owed: account {account} is beyond the seller's subscription capacity from {from}. | +| `excluded:overridden` | not owed | exception | Not owed: another of the caller's delivery configurations takes {account} for this feed from {from}. | +| `excluded:action_required` | not owed | exception | Not owed: the seller serves account {account} only after a pending action completes; excluded from {from}. | +| `gap:unauthorized` | not owed | gap | Not owed: the period falls in the authorization gap from {from} to {until}. | +| `cutover_stretch` | not owed | cutover or timezone | Not owed: the period {start} to {end} in {timezone} lies outside the span its side owes around the handoff at {handoff}. | +| `owed_under_other_version` | not owed | cutover | Not owed under version {v}: the side before the handoff at {handoff} owes this time. | +| `period_not_ended` | not owed | outcome | Not owed yet: the period {start} to {end} has not ended. | +| `owed` | owed | outcome | Owed: {buy} under {offering} for the period {start} to {end}, due at {expected_at}. | +| `owed:regrant_backfill` | owed | outcome | Owed as a re-grant backfill: the period {start} to {end} falls in the interruption from {unauthorized_at} to {regranted_at}, which the seller backfills; due at {expected_at}. | +| `owed:withheld` | owed | outcome | Owed but withheld: the caller lost authority for {account} at {unauthorized_at}, so the period is due again only after a re-grant. | ## Worked traces -The `CoverageTrace` format and its template sentences arrive with `explain` in a later beta. The cases below are among the spec's canonical vectors, which this SDK runs in `tests/test_reporting_subscription_parity.py`. +Each case below is one of the spec's canonical vectors, which this SDK runs in `tests/test_reporting_subscription_parity.py`. The sentences are what `explain` prints for it. ### Confirmation day A period that ends at or before the buy's `confirmed_at` is not owed (`before_confirmation`). The period in which the buy is confirmed ends after it, so it is owed in full. +> Not owed: the period ends 2026-11-04T05:00Z, before mb_123 was confirmed at 2026-11-04T10:00Z. +> +> Owed: mb_123 under daily_pkg for the period 2026-11-04T05:00Z to 2026-11-05T05:00Z, due at 2026-11-05T11:00Z. + ### Cancellation mid-period A buy stops at the earliest of `end_time`, the cancellation's delivery stop and `completed_at`. The delivery stop is `cancellation_effective_at` when the seller delivers through a notice period, else `cancellation.canceled_at` (3.3.0-beta.3). A period that starts at or after the stop is not owed (`after_stop`). The period containing the stop is owed in full. Flight values are read as of each period's end from the buy's `flight_history`, so a retroactive edit never makes an elapsed period owed. +> Owed: mb_1 under off_a_pkg for the period 2026-11-05T00:00Z to 2026-11-06T00:00Z, due at 2026-11-06T06:00Z. +> +> Not owed: the period starts 2026-11-06T00:00Z, at or after mb_1 stopped at 2026-11-05T06:00Z. + ### Re-grant gap A re-grant inside `status_retention_days` continues the chain, so `authorized_at` is kept and `authorization.regranted_at` carries the re-grant. Unless the seller advertises `backfill_on_regrant`, the periods from the last boundary at or before `unauthorized_at` to the first boundary at or after `regranted_at` are a gap (`gap:unauthorized`). Withheld periods are due at `regranted_at` plus `delivery_sla`. +> Not owed: the period falls in the authorization gap from 2026-11-05T06:00Z to 2026-11-06T18:00Z. + +With `backfill_on_regrant`, the same period is `owed:regrant_backfill`, due at the re-grant plus `delivery_sla`. + ### Timezone cutover -A timezone change takes effect at its `effective_at`, and each zone owes only its own span. A previous-zone period starting at or after `effective_at`, or a new-zone period starting before it, is not owed (`cutover_stretch`). +A timezone change takes effect at its `effective_at`, and each zone owes only its own span. A previous-zone period starting at or after `effective_at`, or a new-zone period starting before it, is not owed (`cutover_stretch`, rule `timezone`). + +> Not owed: the period 2026-11-05T05:00Z to 2026-11-06T05:00Z in America/New_York lies outside the span its side owes around the handoff at 2026-11-05T05:00Z. +> +> Owed: mb_1 under off_e_pkg_acct_tz for the period 2026-11-05T08:00Z to 2026-11-06T08:00Z, due at 2026-11-06T14:00Z. ### Billing stub @@ -67,4 +158,6 @@ Coverage resumes at the first boundary at or after the later of the reactivation ### DST -`P1D` periods follow the local calendar, so DST days are 23 or 25 hours long. Under a local alignment each local date's first instant is always a boundary: local midnight, or the first instant after a DST gap at midnight (Havana, Santiago, Cairo). Days therefore nest in every sub-daily schedule, and `PT24H` equals `P1D`. Under `utc`, `PT24H` periods are always 24 hours. A sub-daily `billing_cycle` is refused (`invalid_period_duration`). +> Owed: mb_1 under off_e_pkg_acct_tz for the period 2026-11-01T04:00Z to 2026-11-02T05:00Z, due at 2026-11-02T11:00Z. + +That New York day is 25 hours long. `P1D` periods follow the local calendar, so DST days are 23 or 25 hours long. Under a local alignment each local date's first instant is always a boundary: local midnight, or the first instant after a DST gap at midnight (Havana, Santiago, Cairo). Days therefore nest in every sub-daily schedule, and `PT24H` equals `P1D`. Under `utc`, `PT24H` periods are always 24 hours. A sub-daily `billing_cycle` is refused (`invalid_period_duration`). diff --git a/docs/reporting-subscriptions-seller.md b/docs/reporting-subscriptions-seller.md index b440aa2e9..24f223592 100644 --- a/docs/reporting-subscriptions-seller.md +++ b/docs/reporting-subscriptions-seller.md @@ -99,4 +99,10 @@ Today: `planning_error_handler`, `turn.subscription_errors`, the engine's `evalu ## Troubleshooting with explain -`reporting.explain_coverage(...)` and the shared `CoverageTrace` arrive in a later beta. Until then, `engine.owed_periods(principal, subscription_id, account_id)` lists the owed periods. The coverage reason codes are in [Coverage, explained](reporting-subscriptions-coverage.md#reason-codes). +`await reporting.explain_coverage(principal, account_id, period="2026-11-03")` returns a `CoverageTrace` per period: the verdict, the reason code, a sentence, every rule with the facts it read, and the `reporting_obligation_id` of an owed period in `seller_only`. From a shell: + +``` +python -m adcp reporting explain --app myapp.reporting:service --principal agency-x --account acc_a --period 2026-11-03 +``` + +Pass `invisible_buys=` with buys the caller cannot see through `get_media_buys` to explain them as `buy_not_visible`. `python -m adcp reporting diff buyer.json seller.json` compares a buyer's trace with yours. See [Coverage, explained](reporting-subscriptions-coverage.md#explaining-a-period). diff --git a/src/adcp/__main__.py b/src/adcp/__main__.py index 04a731a58..89c43203c 100644 --- a/src/adcp/__main__.py +++ b/src/adcp/__main__.py @@ -694,6 +694,11 @@ def main() -> None: print(error, file=sys.stderr) sys.exit(2) sys.exit(migrate_main(sys.argv[2:])) + if len(sys.argv) > 1 and sys.argv[1] == "reporting": + # Reporting-subscription tooling: explain coverage, compare traces. + from adcp.reporting.cli import main as reporting_main + + sys.exit(reporting_main(sys.argv[2:])) parser = argparse.ArgumentParser( description="AdCP Client - Interact with AdCP agents", usage="adcp [options] [tool] [payload]", diff --git a/src/adcp/reporting/cli.py b/src/adcp/reporting/cli.py new file mode 100644 index 000000000..402e0e859 --- /dev/null +++ b/src/adcp/reporting/cli.py @@ -0,0 +1,166 @@ +"""``python -m adcp reporting``: explain coverage and compare traces (AdCP 3.3, experimental). + +Seller side, against the seller's own service and storage:: + + python -m adcp reporting explain --app myapp.reporting:service \\ + --principal agency-x --account acc_renault_fr --period 2026-11-03 [--json] + +``--app`` names the seller's :class:`~adcp.reporting.service.ReliableReportingService` +as ``module:attribute`` (a callable attribute is called, and awaited if it is +a coroutine function), the way ASGI servers name an app. ``--tenant`` sets the +principal's tenant. + +Comparing a buyer's trace with a seller's:: + + python -m adcp reporting diff buyer.json seller.json + +prints the first rule whose facts differ, and whether the two sides read the +same inputs (equal ``inputs_digest``: a logic bug) or different state. +""" + +from __future__ import annotations + +import argparse +import asyncio +import importlib +import inspect +import json +import sys +from collections.abc import Sequence +from typing import Any + +USAGE = "python -m adcp reporting {explain,diff} ..." + + +def _load_app(target: str) -> Any: + module_name, _, attribute = target.partition(":") + if not module_name or not attribute: + raise SystemExit("--app is module:attribute, for example myapp.reporting:service") + module = importlib.import_module(module_name) + value: Any = module + for part in attribute.split("."): + value = getattr(value, part) + return value + + +async def _resolve_app(target: str) -> Any: + app = _load_app(target) + if callable(app) and not hasattr(app, "explain_coverage"): + app = app() + if inspect.isawaitable(app): + app = await app + if not hasattr(app, "explain_coverage"): + raise SystemExit(f"{target} is not a ReliableReportingService") + return app + + +def _print_traces(traces: Sequence[Any], *, as_json: bool) -> None: + if as_json: + print(json.dumps([trace.to_dict() for trace in traces], indent=2)) + return + if not traces: + print("No period matches: check the account, the subscription and the period.") + return + for trace in traces: + subject = trace.subject + period = subject["period"] + print( + f"{subject['subscription_id']} v{subject['subscription_version']} " + f"{subject.get('media_buy_id') or '-'} {subject.get('offering_id') or '-'} " + f"{period['start']}..{period['end']} ({period['timezone']}): " + f"{trace.verdict} [{trace.reason}]" + ) + print(f" {trace.explanation}") + + +async def _explain(args: argparse.Namespace) -> int: + from adcp._principal_ref import PrincipalRef + + service = await _resolve_app(args.app) + traces = await service.explain_coverage( + PrincipalRef(args.tenant, args.principal), + args.account, + period=args.period, + subscription_id=args.subscription, + media_buy_id=args.media_buy, + offering_id=args.offering, + ) + _print_traces(traces, as_json=args.json) + return 0 + + +def _read_traces(path: str) -> list[Any]: + from adcp.reporting.subscriptions.explain import CoverageTrace + + with open(path, encoding="utf-8") as handle: + document = json.load(handle) + items = document if isinstance(document, list) else [document] + return [CoverageTrace.from_dict(item) for item in items] + + +def _diff(args: argparse.Namespace) -> int: + from adcp.reporting.subscriptions.explain import diff_coverage_traces + + left, right = _read_traces(args.left), _read_traces(args.right) + + def key(trace: Any) -> tuple[Any, ...]: + subject = trace.subject + return ( + subject["account_id"], + subject["subscription_id"], + subject.get("media_buy_id"), + subject.get("offering_id"), + subject["period"]["start"], + subject["period"]["end"], + ) + + rights = {key(trace): trace for trace in right} + differences = 0 + for trace in left: + other = rights.pop(key(trace), None) + if other is None: + print(f"Only in {args.left}: {key(trace)}") + differences += 1 + continue + found = diff_coverage_traces(trace, other) + if found is not None: + print(f"{key(trace)}: {found.describe()}") + differences += 1 + for missing in rights: + print(f"Only in {args.right}: {missing}") + differences += 1 + if not differences: + print("The traces agree.") + return 1 if differences else 0 + + +def main(argv: Sequence[str]) -> int: + parser = argparse.ArgumentParser(prog="python -m adcp reporting", usage=USAGE) + commands = parser.add_subparsers(dest="command", required=True) + explain = commands.add_parser( + "explain", help="Explain why periods are or are not owed (seller, from your service)" + ) + explain.add_argument("--app", required=True, help="module:attribute of your service") + explain.add_argument("--principal", required=True, help="principal id") + explain.add_argument("--tenant", default=None, help="tenant id, when multi-tenant") + explain.add_argument("--account", required=True, help="account_id") + explain.add_argument( + "--period", + required=True, + help="YYYY-MM-DD read in each offering's zone, or an RFC 3339 instant", + ) + explain.add_argument("--subscription", default=None, help="one subscription_id") + explain.add_argument("--media-buy", default=None, help="one media_buy_id") + explain.add_argument("--offering", default=None, help="one offering_id") + explain.add_argument("--json", action="store_true", help="print the CoverageTrace JSON") + diff = commands.add_parser("diff", help="Compare two CoverageTrace JSON files") + diff.add_argument("left") + diff.add_argument("right") + args = parser.parse_args(list(argv)) + if args.command == "diff": + return _diff(args) + return asyncio.run(_explain(args)) + + +if __name__ == "__main__": # pragma: no cover + sys.exit(main(sys.argv[1:])) diff --git a/src/adcp/reporting/service.py b/src/adcp/reporting/service.py index d1afcbc5f..b8b97f695 100644 --- a/src/adcp/reporting/service.py +++ b/src/adcp/reporting/service.py @@ -808,6 +808,45 @@ async def touch(self, *, account_id: str | None = None, context: Any | None = No managed = self._require_managed() await managed.touch(account_id=account_id, context=context) + async def explain_coverage( + self, + principal: Any, + account_id: str, + *, + period: Any, + subscription_id: str | None = None, + media_buy_id: str | None = None, + offering_id: str | None = None, + invisible_buys: Sequence[Any] = (), + now: datetime | None = None, + ) -> list[Any]: + """Why each selected period is, or is not, owed: the seller's ``CoverageTrace`` list. + + ``period`` is a ``YYYY-MM-DD`` local date read in each offering's + zone, an instant, or ``{"start", "end"}``. Omit ``subscription_id`` + for every subscription covering the account and ``media_buy_id`` for + every buy. Traces carry ``side="seller"`` and a ``seller_only`` block; + call ``trace.for_buyer()`` before showing one to a buyer. See + :func:`adcp.reporting.subscriptions.explain_subscription_coverage`. + """ + if self._managed is not None: + self._compose_managed() + if self._subscriptions is None: + raise ReliableReportingConfigurationError( + "explain_coverage needs ReliableReportingService(subscriptions=...) or " + "reporting_subscriptions=..." + ) + return await self._subscriptions.engine.explain_coverage( + principal, + account_id, + period=period, + subscription_id=subscription_id, + media_buy_id=media_buy_id, + offering_id=offering_id, + invisible_buys=invisible_buys, + now=self._clock() if now is None else now, + ) + @property def grants(self) -> Any: """Push grant changes into the grant journal (``await reporting.grants.changed(...)``). diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index a79aaa0c8..4e8ffb1a7 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -20,6 +20,9 @@ ``principal.changed``, and compute expected periods from its own records. * :class:`ReportingSubscriptionEngine` -- the seller runtime: host hooks, lost authority, exceptions, ``subscription_coverage[]`` and owed periods. +* :func:`explain_subscription_coverage` and :class:`CoverageTrace` -- why a + period is or is not owed, as one JSON document both sides build from the + same wire-visible facts, with the spec's reason codes. """ from __future__ import annotations @@ -78,6 +81,17 @@ decode_principal_state, encode_principal_state, ) +from adcp.reporting.subscriptions.explain import ( + EXPLAIN_TEMPLATES, + TRACE_VERSION, + CoverageReason, + CoverageTrace, + CoverageTraceDifference, + CoverageTraceStep, + coverage_reasons, + diff_coverage_traces, + explain_subscription_coverage, +) from adcp.reporting.subscriptions.facade import ( PreviewResult, ReportingClient, @@ -118,6 +132,15 @@ from adcp.server.reporting_grants import REPORTING_WIND_DOWN_TASKS, wind_down_grant __all__ = [ + "EXPLAIN_TEMPLATES", + "TRACE_VERSION", + "CoverageReason", + "CoverageTrace", + "CoverageTraceDifference", + "CoverageTraceStep", + "coverage_reasons", + "diff_coverage_traces", + "explain_subscription_coverage", "REPORTING_WIND_DOWN_TASKS", "authorization_chain_start", "authorization_regranted_at", diff --git a/src/adcp/reporting/subscriptions/coverage.py b/src/adcp/reporting/subscriptions/coverage.py index f92b470dd..37b9086dd 100644 --- a/src/adcp/reporting/subscriptions/coverage.py +++ b/src/adcp/reporting/subscriptions/coverage.py @@ -888,72 +888,65 @@ def _windows( return subscription_version_windows(subscription, version_configurations=version_configurations) -def owed_subscription_periods( +@dataclass +class _CoverageContext: + """The normalized input one coverage call evaluates: shared by the owed rule and + :mod:`~adcp.reporting.subscriptions.explain`, so both read the same facts.""" + + windows: tuple[SubscriptionVersionWindow, ...] + subscription_id: str + chains: list[_Chain] + changes: list[AccountTimezoneChange] + removals: tuple[SubscriptionException, ...] + package_removals: tuple[SubscriptionException, ...] + membership_intervals: bool + ordered: list[datetime] + selectors: dict[int, SubscriptionSelector] + in_scope: dict[int, bool] + offerings: tuple[SubscriptionOffering, ...] + destinations: Mapping[str, Any] | None + account_timezone: str | None + account_id: str + account_operator: str | None + + def regimes(self, buy: SubscriptionBuy) -> list[_Regime | None]: + return _regimes( + buy, + self.ordered, + self.windows, + self.selectors, + self.in_scope, + self.offerings, + self.destinations, + self.account_timezone, + self.changes, + account_id=self.account_id, + account_operator=self.account_operator, + package_removals=self.package_removals, + ) + + +def _coverage_context( *, subscription: Any, account_id: str, buys: Sequence[SubscriptionBuy], offerings: Sequence[SubscriptionOffering], account_timezone: str | None, - now: datetime, authorizations: Sequence[SubscriptionAuthorization] = (), exceptions: Sequence[Any] = (), gaps: Sequence[Any] = (), coverage: Any | None = None, - backfill_on_regrant: bool = False, status_retention_days: int | None = None, destinations: Mapping[str, Any] | None = None, account_operator: str | None = None, version_configurations: Mapping[int, Any] | None = None, timezone_changes: Sequence[AccountTimezoneChange] = (), - include_unresolved: bool = False, - max_periods: int = DEFAULT_MAX_PERIODS, - max_iterations: int = DEFAULT_MAX_ITERATIONS, -) -> list[OwedSubscriptionPeriod]: - """Every period one subscription owes one account, from plain data. - - Fails closed rather than return a partial set: more than ``max_periods`` - owed (buy, period) pairs, or more than ``max_iterations`` boundaries - examined, raises :class:`CoverageInputError` with code - ``COVERAGE_BUDGET_EXCEEDED`` (the TypeScript SDK's ``max_periods`` and - ``max_iterations``, with the same defaults). - - :param subscription: the ``get_principal`` readback - (``core/reporting-subscription-state.json``) as a model or mapping, or - a sequence of :class:`SubscriptionVersionWindow`. - :param buys: the confirmed buys in the account the caller can see. - A buy without ``confirmed_at`` is not covered. - :param offerings: offerings with the seller's dimension mapping. - :param account_timezone: the account's current timezone, as - ``list_accounts`` shows it. Before a timezone change the zone is that - change's ``previous_timezone``. Required when an ``account_timezone`` - offering resolves without a ``period_timezone`` and no change applies. - :param now: periods are owed once resolved, at their end; a period ending - after ``now`` is returned only with ``include_unresolved``. - :param authorizations: the account's grant history. Empty means the seller - does not advertise ``authorized_at`` and authority never lapsed. - :param exceptions: ``subscription_coverage[].excluded`` entries, or - :class:`SubscriptionException` values. - :param gaps: ``subscription_coverage[].gaps`` entries. - :param coverage: a whole ``subscription_coverage[]`` entry; its - ``excluded``, ``gaps`` and ``timezone_cutovers`` are applied too. - :param backfill_on_regrant: the seller's ``subscriptions.backfill_on_regrant``. - :param status_retention_days: the seller's status retention; required when - a grant was revoked and re-granted. - :param destinations: the caller's ``reporting_destinations`` by - ``destination_id``, needed when a version names a destination. - :param account_operator: the account's operator, for ``operator_domains``. - :param timezone_changes: account timezone changes, each a cutover at its - ``changed_at`` (``timezone_cutovers[].effective_at``). - """ - now = utc_instant(now, field="now") - for name, budget in (("max_periods", max_periods), ("max_iterations", max_iterations)): - if type(budget) is not int or budget < 1: - raise CoverageInputError("invalid_input", f"{name} must be a positive integer") - iterations = pairs = 0 +) -> _CoverageContext | None: + """Validate and normalize a coverage input; ``None`` when there is no version.""" windows = _windows(subscription, version_configurations) if not windows: - return [] + return None subscription_id = _subscription_id(subscription, windows) retention = None if status_retention_days is None else timedelta(days=status_retention_days) chains = _chains(authorizations, retention) @@ -1033,25 +1026,116 @@ def owed_subscription_periods( for window in windows } + return _CoverageContext( + windows=windows, + subscription_id=subscription_id, + chains=chains, + changes=changes, + removals=removals, + package_removals=package_removals, + membership_intervals=membership_intervals, + ordered=ordered, + selectors=selectors, + in_scope=in_scope, + offerings=tuple(offerings), + destinations=destinations, + account_timezone=account_timezone, + account_id=account_id, + account_operator=account_operator, + ) + + +def owed_subscription_periods( + *, + subscription: Any, + account_id: str, + buys: Sequence[SubscriptionBuy], + offerings: Sequence[SubscriptionOffering], + account_timezone: str | None, + now: datetime, + authorizations: Sequence[SubscriptionAuthorization] = (), + exceptions: Sequence[Any] = (), + gaps: Sequence[Any] = (), + coverage: Any | None = None, + backfill_on_regrant: bool = False, + status_retention_days: int | None = None, + destinations: Mapping[str, Any] | None = None, + account_operator: str | None = None, + version_configurations: Mapping[int, Any] | None = None, + timezone_changes: Sequence[AccountTimezoneChange] = (), + include_unresolved: bool = False, + max_periods: int = DEFAULT_MAX_PERIODS, + max_iterations: int = DEFAULT_MAX_ITERATIONS, +) -> list[OwedSubscriptionPeriod]: + """Every period one subscription owes one account, from plain data. + + Fails closed rather than return a partial set: more than ``max_periods`` + owed (buy, period) pairs, or more than ``max_iterations`` boundaries + examined, raises :class:`CoverageInputError` with code + ``COVERAGE_BUDGET_EXCEEDED`` (the TypeScript SDK's ``max_periods`` and + ``max_iterations``, with the same defaults). + + :param subscription: the ``get_principal`` readback + (``core/reporting-subscription-state.json``) as a model or mapping, or + a sequence of :class:`SubscriptionVersionWindow`. + :param buys: the confirmed buys in the account the caller can see. + A buy without ``confirmed_at`` is not covered. + :param offerings: offerings with the seller's dimension mapping. + :param account_timezone: the account's current timezone, as + ``list_accounts`` shows it. Before a timezone change the zone is that + change's ``previous_timezone``. Required when an ``account_timezone`` + offering resolves without a ``period_timezone`` and no change applies. + :param now: periods are owed once resolved, at their end; a period ending + after ``now`` is returned only with ``include_unresolved``. + :param authorizations: the account's grant history. Empty means the seller + does not advertise ``authorized_at`` and authority never lapsed. + :param exceptions: ``subscription_coverage[].excluded`` entries, or + :class:`SubscriptionException` values. + :param gaps: ``subscription_coverage[].gaps`` entries. + :param coverage: a whole ``subscription_coverage[]`` entry; its + ``excluded``, ``gaps`` and ``timezone_cutovers`` are applied too. + :param backfill_on_regrant: the seller's ``subscriptions.backfill_on_regrant``. + :param status_retention_days: the seller's status retention; required when + a grant was revoked and re-granted. + :param destinations: the caller's ``reporting_destinations`` by + ``destination_id``, needed when a version names a destination. + :param account_operator: the account's operator, for ``operator_domains``. + :param timezone_changes: account timezone changes, each a cutover at its + ``changed_at`` (``timezone_cutovers[].effective_at``). + """ + now = utc_instant(now, field="now") + for name, budget in (("max_periods", max_periods), ("max_iterations", max_iterations)): + if type(budget) is not int or budget < 1: + raise CoverageInputError("invalid_input", f"{name} must be a positive integer") + iterations = pairs = 0 + context = _coverage_context( + subscription=subscription, + account_id=account_id, + buys=buys, + offerings=offerings, + account_timezone=account_timezone, + authorizations=authorizations, + exceptions=exceptions, + gaps=gaps, + coverage=coverage, + status_retention_days=status_retention_days, + destinations=destinations, + account_operator=account_operator, + version_configurations=version_configurations, + timezone_changes=timezone_changes, + ) + if context is None: + return [] + subscription_id = context.subscription_id + chains = context.chains + removals = context.removals + owed: dict[tuple[int, str, str, datetime], _Owed] = {} for buy in buys: if buy.confirmed_at is None: continue confirmed_at = buy.confirmed_at - regimes = _regimes( - buy, - ordered, - windows, - selectors, - in_scope, - offerings, - destinations, - account_timezone, - changes, - account_id=account_id, - account_operator=account_operator, - package_removals=package_removals, - ) + regimes = context.regimes(buy) flights_start = buy._earliest_start() latest_stop = buy._latest_stop() for index, regime in enumerate(regimes): diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index e7e584742..ff564a586 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -96,6 +96,7 @@ _MAX_STATE_WRITE_ATTEMPTS = 8 if TYPE_CHECKING: + from adcp.reporting.subscriptions.explain import CoverageTrace from adcp.reporting.subscriptions.ledger import FrozenOwedSet from adcp.server.principal import ( DeliveryConfigIdsHook, @@ -2258,6 +2259,109 @@ def _owed( timezone_changes=list(account.timezone_changes), ) + async def explain_coverage( + self, + identity: PrincipalLike, + account_id: str, + *, + period: Any, + subscription_id: str | None = None, + media_buy_id: str | None = None, + offering_id: str | None = None, + invisible_buys: Sequence[SubscriptionBuy] = (), + now: datetime | None = None, + ) -> list[CoverageTrace]: + """Why each selected period is, or is not, owed: the seller's :class:`CoverageTrace`. + + The same classification as :func:`explain_subscription_coverage` over + the engine's records, the same ones :meth:`owed_periods` reads, so a + trace's verdict is exactly what the engine owes. The ``gaps[]`` the + engine serves enter as the buyer reads them. Every subscription + covering the account is explained unless ``subscription_id`` names + one. ``invisible_buys`` are buys in the account the caller cannot see + through ``get_media_buys`` (made in the seller's UI or by another + agent), explained as ``buy_not_visible``. ``seller_only`` carries + whether the caller can see the buy and, for an owed period, its + ``reporting_obligation_id``; strip it with + :meth:`CoverageTrace.for_buyer` before showing a buyer. + """ + from adcp.reporting.subscriptions.explain import explain_subscription_coverage + from adcp.reporting.subscriptions.ledger import subscription_obligation_id + + instant = self._now(now) + states = [ + _dump(item) + for item in await self._principals.reporting_subscriptions(identity) + if subscription_id is None or item.subscription_id == subscription_id + ] + if subscription_id is not None and not states: + raise KeyError(subscription_id) + destinations = await self._destinations(identity) + buys = list(await self._account_buys(identity, account_id)) + principal = await self._read(identity.subject) + account = principal.accounts.get(account_id) + if account is None: + return [] + traces: list[CoverageTrace] = [] + for state in states: + sub_id = str(state["subscription_id"]) + versions = { + version: configuration + for (key, version), configuration in principal.versions.items() + if key == sub_id + } + windows = tuple( + replace( + window, + destination=principal.destinations.get((sub_id, window.subscription_version)), + ) + for window in subscription_version_windows( + state, version_configurations=versions, require_prior_configurations=True + ) + ) + exclusions = [ + SubscriptionException( + e.reason, e.from_, e.until, e.media_buy_id, package_ids=e.package_ids + ) + for e in principal.exclusions.get((sub_id, account_id), []) + ] + for trace in explain_subscription_coverage( + subscription=windows, + account_id=account_id, + buys=[*buys, *invisible_buys], + invisible_media_buy_ids=[item.media_buy_id for item in invisible_buys], + offerings=self._offerings, + account_timezone=account.timezone, + now=instant, + period=period, + media_buy_id=media_buy_id, + offering_id=offering_id, + side="seller", + principal_id=identity.subject, + authorizations=list(account.grants), + exceptions=exclusions, + gaps=self._gaps(state, account, buys, destinations, instant), + backfill_on_regrant=self._support.backfill_on_regrant, + status_retention_days=self._retention_days, + destinations=destinations, + account_operator=account.operator, + timezone_changes=list(account.timezone_changes), + ): + only: dict[str, Any] = dict(trace.seller_only or {}) + if trace.verdict == "owed": + subject = trace.subject + only["reporting_obligation_id"] = subscription_obligation_id( + account_id=account_id, + principal_id=identity.subject, + subscription_id=sub_id, + subscription_version=int(subject["subscription_version"]), + offering_id=str(subject["offering_id"]), + period_start=_required_instant(subject["period"]["start"]), + period_end=_required_instant(subject["period"]["end"]), + ) + traces.append(replace(trace, seller_only=only)) + return traces + async def frozen_owed_sets( self, identity: PrincipalLike, diff --git a/src/adcp/reporting/subscriptions/explain.py b/src/adcp/reporting/subscriptions/explain.py new file mode 100644 index 000000000..90df9760a --- /dev/null +++ b/src/adcp/reporting/subscriptions/explain.py @@ -0,0 +1,1378 @@ +"""Why one period is, or is not, owed: the shared ``CoverageTrace`` (AdCP 3.3, experimental). + +:func:`explain_subscription_coverage` classifies periods of one subscription on +one account against the same facts :func:`owed_subscription_periods` reads. The +verdict is the owed rule's own: a period is ``owed`` exactly when the rule owes +it. The reason is the first rule, in the spec's fixed order, that rejects the +period, with the facts that rule read. Seller and buyer build the identical +trace from the identical wire-visible facts, so a disagreement between two +traces of the same period is a comparison of two documents: + +* equal ``inputs_digest`` and different verdicts is a logic bug in one SDK; +* different ``inputs_digest`` means one side read stale or different state, + and :func:`diff_coverage_traces` names the first rule whose facts differ. + +The reason codes, their verdicts and their rules are read from the spec's +``reason-codes.json``, packaged with the SDK for the AdCP release it serves. +Explanations are rendered from :data:`EXPLAIN_TEMPLATES`, one sentence per +reason, so both sides emit identical text for identical facts. ``now`` is not +part of the digest: only ``period_not_ended`` depends on it. + +``seller_only`` facts (when the seller froze a period, whether a buy is +visible to the caller, the obligation id) never enter the digest, and a +trace with them is never sent to a buyer: :meth:`CoverageTrace.for_buyer` +strips them. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass, field +from datetime import date, datetime, time, timedelta +from functools import lru_cache +from importlib.resources import files +from typing import Any, Literal + +from adcp.reporting.canonical_json import canonical_json_utf8_v1 +from adcp.reporting.subscriptions._clock import ( + CoverageInputError, + PeriodCalendar, + iso_z, + require_zone, + utc_instant, +) +from adcp.reporting.subscriptions.coverage import ( + DEFAULT_MAX_ITERATIONS, + DEFAULT_MAX_PERIODS, + AccountTimezoneChange, + OwedSubscriptionPeriod, + SubscriptionAuthorization, + SubscriptionBuy, + SubscriptionException, + SubscriptionVersionWindow, + _Chain, + _coverage_context, + _CoverageContext, + _instant, + _Regime, + _timezone_at, + authorization_chain_start, + authorization_regranted_at, + owed_subscription_periods, +) +from adcp.reporting.subscriptions.resolution import SubscriptionOffering +from adcp.reporting.subscriptions.vocabulary import COVERAGE_RULE_ORDER + +#: The ``CoverageTrace`` document version. +TRACE_VERSION = 1 +#: Version of :data:`EXPLAIN_TEMPLATES`. +EXPLAIN_TEMPLATES_VERSION = 1 + +CoverageVerdict = Literal["owed", "not_owed"] +CoverageSide = Literal["buyer", "seller"] +StepResult = Literal["pass", "fail", "not_applicable"] + +#: One sentence per reason code. Placeholders are filled from the trace's +#: subject and the failing step's facts; instants render as RFC 3339 UTC with +#: minute precision when the seconds are zero. Both SDKs ship this table +#: verbatim (proposed for the spec's ``explain-templates.json``). +EXPLAIN_TEMPLATES: Mapping[str, str] = { + "owed": ( + "Owed: {media_buy_id} under {offering_id} for the period {period_start} to " + "{period_end}, due at {expected_at}." + ), + "owed:regrant_backfill": ( + "Owed as a re-grant backfill: the period {period_start} to {period_end} falls in the " + "interruption from {unauthorized_at} to {regranted_at}, which the seller backfills; " + "due at {expected_at}." + ), + "owed:withheld": ( + "Owed but withheld: the caller lost authority for {account_id} at {unauthorized_at}, " + "so the period {period_start} to {period_end} is due again only after a re-grant." + ), + "period_not_ended": ( + "Not owed yet: the period {period_start} to {period_end} has not ended. It becomes " + "owed when it ends unless a later change removes it." + ), + "not_in_scope": ( + "Not owed: version {subscription_version} of {subscription_id} does not select " + "account {account_id}." + ), + "excluded:excluded_by_buyer": ( + "Not owed: the buyer excluded account {account_id} from {subscription_id} from {from}." + ), + "not_authorized": ( + "Not owed: the caller held no authority for {account_id} over the period " + "{period_start} to {period_end}." + ), + "before_coverage_start": ( + "Not owed: the period starts {period_start}, before coverage starts at {coverage_start}." + ), + "subscription_inactive": ( + "Not owed: {subscription_id} was inactive from {deactivated_at} when the period " + "started at {period_start}." + ), + "excluded:offering_unsupported": ( + "Not owed: {offering_id} does not report {media_buy_id} under the selector of " + "version {subscription_version}." + ), + "outside_flight": ( + "Not owed: the period ends {period_end}, at or before {media_buy_id} starts at " + "{start_time}." + ), + "after_stop": ( + "Not owed: the period starts {period_start}, at or after {media_buy_id} stopped at " + "{stop_time}." + ), + "before_confirmation": ( + "Not owed: the period ends {period_end}, before {media_buy_id} was confirmed" + "{confirmed_clause}." + ), + "excluded:capacity": ( + "Not owed: account {account_id} is beyond the seller's subscription capacity from {from}." + ), + "excluded:overridden": ( + "Not owed: another of the caller's delivery configurations takes {account_id} for " + "this feed from {from}." + ), + "excluded:action_required": ( + "Not owed: the seller serves account {account_id} only after a pending action " + "completes; excluded from {from}." + ), + "gap:unauthorized": ( + "Not owed: the period falls in the authorization gap from {from} to {until}." + ), + "cutover_stretch": ( + "Not owed: the period {period_start} to {period_end} in {timezone} lies outside the " + "span its side owes around the handoff at {handoff}." + ), + "owed_under_other_version": ( + "Not owed under version {subscription_version}: the side before the handoff at " + "{handoff} owes this time." + ), + "buy_not_visible": ("Not owed: the caller cannot see {media_buy_id} through get_media_buys."), +} + + +# -- the spec vocabulary -------------------------------------------------------- + + +@dataclass(frozen=True) +class CoverageReason: + """One code of the spec's frozen ``reason-codes.json``.""" + + code: str + verdict: CoverageVerdict + #: The rule that assigns it; ``None`` for the outcome codes. + rule: str | None + #: ``both``, or ``seller`` for a seller-only code. + side: str + definition: str + + +@lru_cache(maxsize=1) +def coverage_reasons() -> Mapping[str, CoverageReason]: + """The reason codes from the spec's ``reason-codes.json`` for the release served. + + Read from the compliance tree packaged with the SDK + (``adcp/_compliance//test-vectors/reporting-subscriptions/``). + """ + release = files("adcp").joinpath("ADCP_VERSION").read_text(encoding="utf-8").strip() + path = files("adcp").joinpath( + f"_compliance/{release}/test-vectors/reporting-subscriptions/reason-codes.json" + ) + document = json.loads(path.read_text(encoding="utf-8")) + if tuple(document["rule_order"]) != COVERAGE_RULE_ORDER: + raise RuntimeError("reason-codes.json rule_order differs from COVERAGE_RULE_ORDER") + return { + item["code"]: CoverageReason( + code=item["code"], + verdict=item["verdict"], + rule=item.get("rule"), + side=item["side"], + definition=item["definition"], + ) + for item in document["reasons"] + } + + +def _reason(code: str) -> CoverageReason: + reason = coverage_reasons().get(code) + if reason is None: # pragma: no cover - a vocabulary bug, never input + raise RuntimeError(f"{code!r} is not a spec reason code") + return reason + + +# -- the trace ------------------------------------------------------------------- + + +@dataclass(frozen=True) +class CoverageTraceStep: + """One rule as applied to the period, with the facts it read.""" + + rule: str + result: StepResult + facts: Mapping[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return {"rule": self.rule, "result": self.result, "facts": dict(self.facts)} + + +@dataclass(frozen=True) +class CoverageTrace: + """Why one period of one (buy, offering) under one subscription is or is not owed. + + The same JSON in both SDKs and on both sides of the wire + (:meth:`to_dict`). ``seller_only`` is never sent to a buyer. + """ + + side: CoverageSide + subject: Mapping[str, Any] + verdict: CoverageVerdict + reason: str + explanation: str + steps: tuple[CoverageTraceStep, ...] + inputs_digest: str + seller_only: Mapping[str, Any] | None = None + trace_version: int = TRACE_VERSION + + @property + def rule(self) -> str | None: + """The rule that assigned :attr:`reason`; ``None`` for an outcome code.""" + failed = next((step.rule for step in self.steps if step.result == "fail"), None) + return failed if failed is not None else _reason(self.reason).rule + + def for_buyer(self) -> CoverageTrace: + """This trace without ``seller_only``, the form that may leave the seller.""" + return CoverageTrace( + side=self.side, + subject=self.subject, + verdict=self.verdict, + reason=self.reason, + explanation=self.explanation, + steps=self.steps, + inputs_digest=self.inputs_digest, + trace_version=self.trace_version, + ) + + def to_dict(self) -> dict[str, Any]: + result: dict[str, Any] = { + "trace_version": self.trace_version, + "side": self.side, + "subject": _plain(self.subject), + "verdict": self.verdict, + "reason": self.reason, + "explanation": self.explanation, + "steps": [step.to_dict() for step in self.steps], + "inputs_digest": self.inputs_digest, + } + if self.seller_only is not None: + result["seller_only"] = _plain(self.seller_only) + return result + + def to_json(self) -> str: + return json.dumps(self.to_dict(), indent=2, sort_keys=False) + + @classmethod + def from_dict(cls, value: Mapping[str, Any]) -> CoverageTrace: + if value.get("trace_version") != TRACE_VERSION: + raise ValueError(f"unsupported trace_version {value.get('trace_version')!r}") + return cls( + side=value["side"], + subject=dict(value["subject"]), + verdict=value["verdict"], + reason=value["reason"], + explanation=value["explanation"], + steps=tuple( + CoverageTraceStep(item["rule"], item["result"], dict(item.get("facts") or {})) + for item in value["steps"] + ), + inputs_digest=value["inputs_digest"], + seller_only=value.get("seller_only"), + ) + + +@dataclass(frozen=True) +class CoverageTraceDifference: + """Where two traces of the same period first disagree.""" + + field: str + #: The first rule whose result or facts differ, when the difference is in a step. + rule: str | None + left: Any + right: Any + #: Whether the digests are equal: then the difference is a logic bug in one SDK, + #: otherwise one side read different state. + same_inputs: bool + + def describe(self) -> str: + where = f"rule {self.rule!r}" if self.rule else self.field + cause = ( + "Equal inputs_digest: one SDK applied the rule differently." + if self.same_inputs + else "Different inputs_digest: the two sides read different state." + ) + return f"First difference at {where}: {self.left!r} != {self.right!r}. {cause}" + + +def diff_coverage_traces( + left: CoverageTrace, right: CoverageTrace +) -> CoverageTraceDifference | None: + """The first difference between two traces of one period, or ``None``. + + Compares the subject, then each rule's result and facts in rule order, + then the verdict and reason. ``side``, ``explanation`` wording beyond the + reason, and ``seller_only`` are not compared. + """ + same = left.inputs_digest == right.inputs_digest + # principal_id is the seller's name for the caller; a buyer trace omits it. + subjects = [ + {key: value for key, value in _plain(trace.subject).items() if key != "principal_id"} + for trace in (left, right) + ] + if subjects[0] != subjects[1]: + return CoverageTraceDifference("subject", None, subjects[0], subjects[1], same) + lefts = {step.rule: step for step in left.steps} + rights = {step.rule: step for step in right.steps} + for rule in COVERAGE_RULE_ORDER: + a, b = lefts.get(rule), rights.get(rule) + if a is None and b is None: + continue + a_doc = None if a is None else a.to_dict() + b_doc = None if b is None else b.to_dict() + if _plain(a_doc) != _plain(b_doc): + return CoverageTraceDifference("steps", rule, a_doc, b_doc, same) + for name in ("verdict", "reason"): + if getattr(left, name) != getattr(right, name): + return CoverageTraceDifference( + name, None, getattr(left, name), getattr(right, name), same + ) + if not same: + return CoverageTraceDifference( + "inputs_digest", None, left.inputs_digest, right.inputs_digest, same + ) + return None + + +# -- helpers --------------------------------------------------------------------- + + +def _plain(value: Any) -> Any: + return json.loads(json.dumps(value, default=_json_default, sort_keys=True)) + + +def _json_default(value: Any) -> Any: + if isinstance(value, datetime): + return iso_z(value) + if isinstance(value, (set, frozenset, tuple)): + return sorted(value) if isinstance(value, (set, frozenset)) else list(value) + if hasattr(value, "model_dump"): + return value.model_dump(mode="json", exclude_none=True) + raise TypeError(f"{type(value).__name__} is not JSON serializable") + + +def _z(value: datetime | None) -> str | None: + return None if value is None else iso_z(value) + + +def _text_instant(value: Any) -> str: + """An instant for an explanation: RFC 3339 UTC, minutes when the seconds are zero.""" + if value is None: + return "an unknown time" + if isinstance(value, str): + instant = _instant(value, field="explanation") + if instant is None: # pragma: no cover + return value + value = instant + instant = utc_instant(value) + if instant.second == 0 and instant.microsecond == 0: + return instant.strftime("%Y-%m-%dT%H:%MZ") + return iso_z(instant) + + +def _dump(value: Any) -> dict[str, Any]: + if hasattr(value, "model_dump"): + dumped = value.model_dump(mode="json", exclude_none=True, by_alias=True) + assert isinstance(dumped, dict) + return dumped + return {key: item for key, item in dict(value).items() if item is not None} + + +_SET_FIELDS = ("account_ids", "operator_domains") +_RFC3339 = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}") + + +def _normalized_configuration(configuration: Any) -> dict[str, Any]: + """A version's configuration as both sides read it: sets sorted, defaults spelled out.""" + payload = { + key: value + for key, value in _dump(configuration).items() + if key not in {"active", "replace_account_configs", "context", "ext"} + } + scope = dict(payload.get("scope") or {}) + if "account_ids" in scope: + scope["account_ids"] = sorted(scope["account_ids"]) + payload["scope"] = scope + for name in ("exclude_account_ids", "operator_domains"): + if name in payload: + payload[name] = sorted(payload[name]) + if "destination_overrides" in payload: + payload["destination_overrides"] = sorted( + ( + {**item, **{name: sorted(item[name]) for name in _SET_FIELDS if name in item}} + for item in payload["destination_overrides"] + ), + key=lambda item: str(item.get("destination_id")), + ) + selector = dict(payload.get("selector") or {}) + if "dimensions" in selector: + selector["dimensions"] = sorted(selector["dimensions"]) + selector.setdefault("match", "exact" if "dimensions" in selector else "finest") + payload["selector"] = selector + return payload + + +def _authorization_projection( + authorizations: Sequence[SubscriptionAuthorization], retention_days: int | None +) -> dict[str, Any]: + """The wire-visible authorization: chain start, latest re-grant, and a current loss.""" + if not authorizations: + return {} + retention = 10**6 if retention_days is None else retention_days + last = authorizations[-1] + if last.unauthorized_at is None: + return { + "authorized_at": _z( + authorization_chain_start(authorizations, status_retention_days=retention) + ), + "regranted_at": _z( + authorization_regranted_at(authorizations, status_retention_days=retention) + ), + } + open_history = [*authorizations[:-1], SubscriptionAuthorization(last.authorized_at)] + return { + "authorized_at": _z( + authorization_chain_start(open_history, status_retention_days=retention) + ), + "regranted_at": _z( + authorization_regranted_at(open_history, status_retention_days=retention) + ), + "unauthorized_at": _z(last.unauthorized_at), + } + + +def _offering_facts(offering: SubscriptionOffering | None) -> dict[str, Any] | None: + if offering is None: + return None + return { + "offering_id": offering.offering_id, + "report_definition_id": offering.report_definition_id, + "feed_purpose": offering.feed_purpose, + "period_duration": offering.period_duration, + "alignment": offering.alignment, + "delivery_sla": offering.delivery_sla, + "period_timezone": offering.period_timezone, + "period_anchor": _z(offering.period_anchor), + "available_from": _z(offering.available_from), + "withdrawn_at": _z(offering.withdrawn_at), + } + + +def _buy_facts(buy: SubscriptionBuy) -> dict[str, Any]: + return { + "media_buy_id": buy.media_buy_id, + "confirmed_at": _z(buy.confirmed_at), + "start_time": _z(buy.start_time), + "end_time": _z(buy.end_time), + "canceled_at": _z(buy.canceled_at), + "cancellation_effective_at": _z(buy.cancellation_effective_at), + "completed_at": _z(buy.completed_at), + "packages": [ + {"package_id": package_id, "reporting_delivery_offering_ids": sorted(ids)} + for package_id, ids in buy.product_offering_ids() + ], + "flight_history": [ + { + "superseded_at": _z(flight.superseded_at), + "start_time": _z(flight.start_time), + "stop_time": _z(flight.stop_time), + } + for flight in buy.superseded_flights + ], + } + + +def _exception_facts(item: SubscriptionException) -> dict[str, Any]: + reason = "gap" if item.reason == "unauthorized" else item.reason + facts: dict[str, Any] = {"reason": reason, "from": _z(item.from_), "until": _z(item.until)} + if item.media_buy_id is not None: + facts["media_buy_id"] = item.media_buy_id + if item.offering_id is not None: + facts["offering_id"] = item.offering_id + if item.package_ids: + facts["package_ids"] = sorted(item.package_ids) + return facts + + +def _window_at( + windows: Sequence[SubscriptionVersionWindow], instant: datetime +) -> SubscriptionVersionWindow: + """The version whose ``[accepted_at, ends_at)`` holds ``instant``, else the nearest one.""" + for window in windows: + if window.accepted_at <= instant and (window.ends_at is None or instant < window.ends_at): + return window + if instant < windows[0].accepted_at: + return windows[0] + return windows[-1] + + +# -- period selection ------------------------------------------------------------ + + +@dataclass(frozen=True) +class _Candidate: + buy: SubscriptionBuy + offering: SubscriptionOffering | None + calendar: PeriodCalendar | None + start: datetime + end: datetime + zone: str + + +def _period_spec(period: Any) -> tuple[str, Any]: + """``("date", date)``, ``("instant", datetime)`` or ``("range", (start, end))``.""" + if isinstance(period, date) and not isinstance(period, datetime): + return "date", period + if isinstance(period, datetime): + return "instant", utc_instant(period, field="period") + if isinstance(period, str): + if len(period) == 10: + try: + return "date", date.fromisoformat(period) + except ValueError as error: + raise CoverageInputError( + "invalid_input", f"period {period!r} is not a YYYY-MM-DD date" + ) from error + instant = _instant(period, field="period") + assert instant is not None + return "instant", instant + if isinstance(period, Mapping): + start = _instant(period.get("start"), field="period.start") + end = _instant(period.get("end"), field="period.end") + if start is None or end is None or end <= start: + raise CoverageInputError("invalid_input", "period needs a start before its end") + return "range", (start, end) + if isinstance(period, Sequence) and len(period) == 2: + return _period_spec({"start": period[0], "end": period[1]}) + raise CoverageInputError( + "invalid_input", + "period is a YYYY-MM-DD date read in the offering's zone, an instant, or {start, end}", + ) + + +def _local_day(day: date, zone: str) -> tuple[datetime, datetime]: + tz = require_zone(zone) + start = datetime.combine(day, time(0), tz) + end = datetime.combine(day + timedelta(days=1), time(0), tz) + return utc_instant(start), utc_instant(end) + + +def _zones( + context: _CoverageContext, offering: SubscriptionOffering, around: Iterable[datetime] +) -> list[str]: + """Every zone the offering's periods can take: both sides of a timezone cutover, and + each version's ``period_timezone`` override.""" + overrides = list(dict.fromkeys(s.period_timezone for s in context.selectors.values())) + accounts = [context.account_timezone] + for change in context.changes: + if change.offering_id in {None, offering.offering_id}: + accounts.extend((change.previous_timezone, change.timezone)) + for instant in around: + accounts.append( + _timezone_at(instant, context.account_timezone, context.changes, offering.offering_id) + ) + zones: list[str] = [] + for override in overrides: + for zone in accounts: + try: + name = offering.zone(zone, override) + except CoverageInputError: + continue + if name not in zones: + zones.append(name) + if not zones: + offering.zone(context.account_timezone, None) # raises the input error + return zones + + +def _periods_for( + calendar: PeriodCalendar, kind: str, value: Any, zone: str +) -> list[tuple[datetime, datetime]]: + if kind == "range": + start, end = value + if ( + calendar.floor(start) == start + and calendar.ceil(start + timedelta(microseconds=1)) == end + ): + return [(start, end)] + return [] + if kind == "instant": + start = calendar.floor(value) + return list( + calendar.periods(starts_from=start, starts_before=start + timedelta(microseconds=1)) + ) + day_start, day_end = _local_day(value, zone) + return list(calendar.periods(starts_from=calendar.ceil(day_start), starts_before=day_end)) + + +# -- the classifier -------------------------------------------------------------- + + +@dataclass +class _Walk: + steps: list[CoverageTraceStep] = field(default_factory=list) + reason: str | None = None + params: dict[str, Any] = field(default_factory=dict) + + def passed(self, rule: str, facts: Mapping[str, Any] | None = None) -> None: + self.steps.append(CoverageTraceStep(rule, "pass", dict(facts or {}))) + + def skipped(self, rule: str, facts: Mapping[str, Any] | None = None) -> None: + self.steps.append(CoverageTraceStep(rule, "not_applicable", dict(facts or {}))) + + def failed(self, rule: str, reason: str, facts: Mapping[str, Any], **params: Any) -> None: + self.steps.append(CoverageTraceStep(rule, "fail", dict(facts))) + self.reason = reason + self.params.update(params) + + +def _regime_at(regimes: Sequence[_Regime | None], instant: datetime) -> _Regime | None: + for regime in regimes: + if regime is None: + continue + if regime.start <= instant and (regime.until is None or instant < regime.until): + return regime + return None + + +def _chain_facts(chains: Sequence[_Chain]) -> dict[str, Any]: + if not chains: + return {"authorized_at": None, "chain": "not_advertised"} + chain = chains[-1] + regrants = [regranted for _lost, regranted in chain.interruptions if regranted is not None] + return { + "authorized_at": _z(chain.start), + "unauthorized_at": _z(chain.end), + "regranted_at": _z(max(regrants)) if regrants else None, + "chain": "continued" if chain.interruptions else ("new" if len(chains) > 1 else "single"), + } + + +def _interruption( + chains: Sequence[_Chain], start: datetime, end: datetime +) -> tuple[datetime, datetime] | None: + for chain in chains: + for lost, regranted in chain.interruptions: + if regranted is not None and end > lost and start < regranted: + return lost, regranted + return None + + +def _classify( + context: _CoverageContext, + candidate: _Candidate, + owed: OwedSubscriptionPeriod | None, + owed_elsewhere: OwedSubscriptionPeriod | None, + *, + now: datetime, + backfill: bool, + invisible: bool, +) -> tuple[_Walk, int]: + """The rule walk for one candidate period. Returns the walk and the subject version.""" + walk = _Walk() + buy, offering, start, end = candidate.buy, candidate.offering, candidate.start, candidate.end + window = _window_at(context.windows, start) + version = owed.delivery_config_version if owed is not None else window.subscription_version + configuration = _dump(window.configuration) + scope = configuration.get("scope") or {} + + # scope ----------------------------------------------------------------- + scope_facts: dict[str, Any] = { + "scope": "account_ids" if "account_ids" in scope else "all_authorized_accounts", + "operator_domains": sorted(configuration.get("operator_domains") or []) or None, + "operator": context.account_operator, + } + buyer_exclusion = next( + ( + item + for item in context.removals + if item.reason == "excluded_by_buyer" and item.removes(start, buy.media_buy_id) + ), + None, + ) + excluded_ids = {str(item) for item in configuration.get("exclude_account_ids") or ()} + if invisible: + walk.failed("scope", "buy_not_visible", {**scope_facts, "buy_visible": False}) + return walk, version + if buyer_exclusion is not None or ( + not context.membership_intervals and context.account_id in excluded_ids + ): + facts = {**scope_facts, "excluded_by_buyer": True} + if buyer_exclusion is not None: + facts.update(_exception_facts(buyer_exclusion)) + walk.failed( + "scope", + "excluded:excluded_by_buyer", + facts, + **{"from": buyer_exclusion.from_ if buyer_exclusion else window.accepted_at}, + ) + return walk, version + if not context.in_scope[window.subscription_version]: + walk.failed("scope", "not_in_scope", scope_facts) + return walk, version + walk.passed("scope", scope_facts) + + # authorization --------------------------------------------------------- + chains = context.chains + authorization_facts = _chain_facts(chains) + if chains and start >= chains[0].start: + held = any( + start >= chain.start and (chain.end is None or end <= chain.end) for chain in chains + ) + if not held: + lost = next((c for c in chains if c.end is not None and end > c.end), None) + walk.failed( + "authorization", + "not_authorized", + authorization_facts, + unauthorized_at=None if lost is None else lost.end, + ) + return walk, version + walk.passed("authorization", authorization_facts) + + # version --------------------------------------------------------------- + selecting = [w.accepted_at for w in context.windows if context.in_scope[w.subscription_version]] + coverage_start = max([min(selecting), *([chains[0].start] if chains else [])]) + if candidate.calendar is not None: + coverage_start = candidate.calendar.ceil(coverage_start) + version_facts = { + "subscription_version": window.subscription_version, + "accepted_at": _z(window.accepted_at), + "coverage_start": _z(coverage_start), + } + if start < coverage_start: + walk.failed( + "version", "before_coverage_start", version_facts, coverage_start=coverage_start + ) + return walk, version + walk.passed("version", version_facts) + + # activity -------------------------------------------------------------- + inactive = [ + {"from": _z(begin), "until": _z(until)} for begin, until in window.inactive_intervals + ] + activity_facts: dict[str, Any] = {"inactive_intervals": inactive} + interval = next( + ( + (begin, until) + for begin, until in window.inactive_intervals + if begin <= start and (until is None or start < until) + ), + None, + ) + deactivated = ( + window is context.windows[-1] and window.ends_at is not None and start >= window.ends_at + ) + if interval is not None or deactivated: + deactivated_at = interval[0] if interval is not None else window.ends_at + walk.failed( + "activity", + "subscription_inactive", + {**activity_facts, "deactivated_at": _z(deactivated_at)}, + deactivated_at=deactivated_at, + ) + return walk, version + walk.passed("activity", activity_facts) + + # offering -------------------------------------------------------------- + regimes = context.regimes(buy) if buy.confirmed_at is not None else [] + regime = _regime_at(regimes, start) + candidates = sorted({oid for _pkg, ids in buy.product_offering_ids() for oid in ids}) + resolved = [] if regime is None else [item.offering.offering_id for item in regime.resolved] + selector = context.selectors[window.subscription_version] + offering_facts = { + "match": selector.match, + "candidates": candidates, + "resolved": resolved, + "selected": None if offering is None else offering.offering_id, + } + unsupported = next( + ( + item + for item in context.removals + if item.reason == "offering_unsupported" + and item.removes( + start, buy.media_buy_id, None if offering is None else offering.offering_id + ) + ), + None, + ) + if unsupported is not None: + walk.failed( + "offering", + "excluded:offering_unsupported", + {**offering_facts, **_exception_facts(unsupported)}, + ) + return walk, version + resolves = offering is not None and ( + offering.offering_id in resolved + # An unconfirmed buy resolves nothing yet; judge it by confirmation. + or (buy.confirmed_at is None and offering.offering_id in candidates) + ) + if not resolves: + walk.failed("offering", "excluded:offering_unsupported", offering_facts) + return walk, version + walk.passed("offering", offering_facts) + + # flight ---------------------------------------------------------------- + flight_start, flight_stop = buy.flight_at(end) + flight_facts = { + "start_time": _z(flight_start), + "stop_time": _z(flight_stop), + "as_of": _z(end), + } + if flight_stop <= flight_start or end <= flight_start: + walk.failed("flight", "outside_flight", flight_facts, start_time=flight_start) + return walk, version + if start >= flight_stop: + walk.failed("flight", "after_stop", flight_facts, stop_time=flight_stop) + return walk, version + walk.passed("flight", flight_facts) + + # confirmation ---------------------------------------------------------- + confirmation_facts = {"confirmed_at": _z(buy.confirmed_at)} + if buy.confirmed_at is None or end <= buy.confirmed_at: + walk.failed( + "confirmation", + "before_confirmation", + confirmation_facts, + confirmed_clause=( + f" at {_text_instant(buy.confirmed_at)}" + if buy.confirmed_at + else " (it is not confirmed yet)" + ), + ) + return walk, version + walk.passed("confirmation", confirmation_facts) + + # exception ------------------------------------------------------------- + exception = next( + ( + item + for item in context.removals + if item.reason in {"capacity", "overridden", "action_required"} + and item.removes(start, buy.media_buy_id, offering.offering_id if offering else None) + ), + None, + ) + if exception is not None: + walk.failed( + "exception", + f"excluded:{exception.reason}", + _exception_facts(exception), + **{"from": exception.from_, "until": exception.until}, + ) + return walk, version + walk.passed("exception", {"open": []}) + + # gap ------------------------------------------------------------------- + gap = next( + ( + item + for item in context.removals + if item.reason == "unauthorized" + and item.removes(start, buy.media_buy_id, offering.offering_id if offering else None) + ), + None, + ) + interruption = _interruption(chains, start, end) + if gap is not None: + walk.failed( + "gap", + "gap:unauthorized", + _exception_facts(gap), + **{"from": gap.from_, "until": gap.until}, + ) + return walk, version + gap_facts = { + "backfill_on_regrant": backfill, + "interruption": None + if interruption is None + else {"unauthorized_at": _z(interruption[0]), "regranted_at": _z(interruption[1])}, + } + if interruption is not None and not backfill: + walk.failed( + "gap", + "gap:unauthorized", + gap_facts, + **{"from": interruption[0], "until": interruption[1]}, + ) + return walk, version + walk.passed("gap", gap_facts) + + # cutover and timezone -------------------------------------------------- + handoff = None + if regime is not None and offering is not None: + handoff = regime.handoffs.get(offering.offering_id) + cutover_facts: dict[str, Any] = {"handoff": _z(handoff)} + timezone_rule = offering is not None and offering.alignment == "account_timezone" + zone_now = ( + _timezone_at(start, context.account_timezone, context.changes, offering.offering_id) + if offering is not None + else None + ) + zone_facts = {"timezone": candidate.zone, "zone_in_effect": zone_now} + if owed is None: + if timezone_rule and selector.period_timezone is None and zone_now != candidate.zone: + walk.passed("cutover", cutover_facts) + walk.failed( + "timezone", + "cutover_stretch", + zone_facts, + handoff=next( + ( + change.changed_at + for change in context.changes + if change.offering_id in {None, offering.offering_id if offering else None} + ), + None, + ), + timezone=candidate.zone, + ) + return walk, version + straddles = handoff is not None and start < handoff < end + reason = ( + "cutover_stretch" if straddles or owed_elsewhere is None else "owed_under_other_version" + ) + if owed_elsewhere is not None: + cutover_facts["owed_under_version"] = owed_elsewhere.delivery_config_version + walk.failed("cutover", reason, cutover_facts, handoff=handoff, timezone=candidate.zone) + return walk, version + walk.passed("cutover", cutover_facts) + if timezone_rule: + walk.passed("timezone", zone_facts) + else: + walk.skipped("timezone", zone_facts) + + # outcome --------------------------------------------------------------- + walk.params.update(expected_at=owed.expected_at) + if end > now: + walk.reason = "period_not_ended" + elif owed.withheld: + last = chains[-1] if chains else None + walk.reason = "owed:withheld" + walk.params.update(unauthorized_at=None if last is None else last.end) + elif backfill and interruption is not None: + walk.reason = "owed:regrant_backfill" + walk.params.update(unauthorized_at=interruption[0], regranted_at=interruption[1]) + else: + walk.reason = "owed" + return walk, version + + +class _Defaults(dict[str, Any]): + """Template values; a fact the walk did not read renders as unknown.""" + + def __missing__(self, key: str) -> str: + return "an unknown time" + + +def _explanation(reason: str, subject: Mapping[str, Any], params: Mapping[str, Any]) -> str: + period = subject["period"] + values: dict[str, Any] = { + "account_id": subject["account_id"], + "subscription_id": subject["subscription_id"], + "subscription_version": subject["subscription_version"], + "media_buy_id": subject.get("media_buy_id") or "the buy", + "offering_id": subject.get("offering_id") or "no offering", + "period_start": _text_instant(period["start"]), + "period_end": _text_instant(period["end"]), + "timezone": period.get("timezone") or "UTC", + } + for key, value in params.items(): + if isinstance(value, datetime) or (isinstance(value, str) and _RFC3339.match(value)): + values[key] = _text_instant(value) + elif value is None: + values[key] = "an unknown time" + else: + values[key] = value + if params.get("timezone"): + values["timezone"] = params["timezone"] + + return EXPLAIN_TEMPLATES[reason].format_map(_Defaults(values)) + + +def _inputs_document( + context: _CoverageContext, + candidate: _Candidate, + subject: Mapping[str, Any], + *, + authorization: Mapping[str, Any], + backfill: bool, + status_retention_days: int | None, +) -> dict[str, Any]: + """The wire-visible facts one period's classification depends on. + + Local to the subject, and in the form both sides can build from what the + wire carries: the version in effect at the period start, the account's + authorization as ``list_accounts`` and ``accounts_unauthorized`` show it, + the zone in effect, the buy as ``get_media_buys`` returns it, the + offering, and the ``excluded[]`` entries covering the period start. An + authorization gap enters as whether it covers the period, which the + seller's ``gaps[]`` and its grant journal state identically. + """ + start, end = candidate.start, candidate.end + window = _window_at(context.windows, start) + buy_id = candidate.buy.media_buy_id + offering_id = candidate.offering.offering_id if candidate.offering else None + covering = sorted( + ( + _exception_facts(item) + for item in context.removals + if item.reason != "unauthorized" and item.removes(start, buy_id, offering_id) + ), + key=lambda item: json.dumps(item, sort_keys=True), + ) + packages = sorted( + ( + _exception_facts(item) + for item in context.package_removals + if item.media_buy_id in {None, buy_id} + and item.from_ <= start + and (item.until is None or start < item.until) + ), + key=lambda item: json.dumps(item, sort_keys=True), + ) + gap = any( + item.reason == "unauthorized" and item.removes(start, buy_id, offering_id) + for item in context.removals + ) or (not backfill and _interruption(context.chains, start, end) is not None) + zone = None + if candidate.offering is not None: + zone = _timezone_at( + start, context.account_timezone, context.changes, candidate.offering.offering_id + ) + return { + "trace_version": TRACE_VERSION, + "subject": {k: v for k, v in subject.items() if k != "principal_id"}, + "version": { + "subscription_version": window.subscription_version, + "accepted_at": _z(window.accepted_at), + "ends_at": _z(window.ends_at), + "inactive_intervals": [[_z(a), _z(b)] for a, b in window.inactive_intervals], + "configuration": _normalized_configuration(window.configuration), + }, + "first_selecting_accepted_at": _z( + min( + ( + w.accepted_at + for w in context.windows + if context.in_scope[w.subscription_version] + ), + default=None, + ) + ), + "account": {"operator": context.account_operator, "zone_in_effect": zone}, + "authorization": dict(authorization), + "buy": _buy_facts(candidate.buy), + "offering": _offering_facts(candidate.offering), + "exceptions": covering, + "package_exceptions": packages, + "authorization_gap": gap, + "backfill_on_regrant": backfill, + "status_retention_days": status_retention_days, + } + + +def _digest(document: Mapping[str, Any]) -> str: + return "sha256:" + hashlib.sha256(canonical_json_utf8_v1(_plain(document))).hexdigest() + + +def explain_subscription_coverage( + *, + subscription: Any, + account_id: str, + buys: Sequence[SubscriptionBuy], + offerings: Sequence[SubscriptionOffering], + account_timezone: str | None, + now: datetime, + period: Any, + media_buy_id: str | None = None, + offering_id: str | None = None, + side: CoverageSide = "buyer", + principal_id: str | None = None, + invisible_media_buy_ids: Iterable[str] = (), + seller_only: Mapping[str, Any] | None = None, + authorizations: Sequence[SubscriptionAuthorization] = (), + exceptions: Sequence[Any] = (), + gaps: Sequence[Any] = (), + coverage: Any | None = None, + backfill_on_regrant: bool = False, + status_retention_days: int | None = None, + destinations: Mapping[str, Any] | None = None, + account_operator: str | None = None, + version_configurations: Mapping[int, Any] | None = None, + timezone_changes: Sequence[AccountTimezoneChange] = (), + max_periods: int = DEFAULT_MAX_PERIODS, + max_iterations: int = DEFAULT_MAX_ITERATIONS, +) -> list[CoverageTrace]: + """One :class:`CoverageTrace` per (buy, offering, period) the request selects. + + Takes exactly the input of :func:`owed_subscription_periods`, plus: + + :param period: a ``YYYY-MM-DD`` local date, read in each offering's zone + (every period of the offering that starts on that date); an instant + (the period containing it); or ``{"start", "end"}``, which must be a + period of the offering's schedule. + :param media_buy_id: one buy; every buy the caller can see when omitted. + :param offering_id: one offering; every offering the buy's packages list + when omitted. A buy that resolves no offering yields one trace without + ``offering_id``. + :param side: ``buyer`` or ``seller``; only the seller passes + ``invisible_media_buy_ids`` (buys in the account the caller cannot see, + reason ``buy_not_visible``) and ``seller_only``. + + Traces are sorted by period start, buy and offering (byte order). A buy + without a match for the selection yields no trace. + """ + now = utc_instant(now, field="now") + if side not in ("buyer", "seller"): + raise ValueError("side is 'buyer' or 'seller'") + hidden = frozenset(invisible_media_buy_ids) + if hidden and side != "seller": + raise ValueError("invisible_media_buy_ids is seller-only") + every_buy = list(buys) + context = _coverage_context( + subscription=subscription, + account_id=account_id, + buys=every_buy, + offerings=offerings, + account_timezone=account_timezone, + authorizations=authorizations, + exceptions=exceptions, + gaps=gaps, + coverage=coverage, + status_retention_days=status_retention_days, + destinations=destinations, + account_operator=account_operator, + version_configurations=version_configurations, + timezone_changes=timezone_changes, + ) + if context is None: + return [] + visible = [buy for buy in every_buy if buy.media_buy_id not in hidden] + owed = owed_subscription_periods( + subscription=context.windows, + account_id=account_id, + buys=visible, + offerings=offerings, + account_timezone=account_timezone, + now=now, + authorizations=authorizations, + exceptions=exceptions, + gaps=gaps, + coverage=coverage, + backfill_on_regrant=backfill_on_regrant, + status_retention_days=status_retention_days, + destinations=destinations, + account_operator=account_operator, + timezone_changes=timezone_changes, + include_unresolved=True, + max_periods=max_periods, + max_iterations=max_iterations, + ) + owed_index: dict[tuple[str, str, datetime, datetime, str], OwedSubscriptionPeriod] = {} + for item in owed: + start = utc_instant(datetime.fromisoformat(item.period_start.replace("Z", "+00:00"))) + end = utc_instant(datetime.fromisoformat(item.period_end.replace("Z", "+00:00"))) + for buy_id in item.media_buy_ids: + owed_index[(buy_id, item.offering_id, start, end, item.source_timezone)] = item + + kind, value = _period_spec(period) + selected = [ + buy for buy in every_buy if media_buy_id is None or buy.media_buy_id == media_buy_id + ] + if media_buy_id is not None and not selected: + raise CoverageInputError( + "invalid_input", f"media_buy_id {media_buy_id!r} is not among the buys passed" + ) + by_id = {item.offering_id: item for item in offerings} + if offering_id is not None and offering_id not in by_id: + raise CoverageInputError( + "invalid_input", f"offering_id {offering_id!r} is not among the offerings passed" + ) + stubs = { + (buy_id, item.offering_id): ( + utc_instant(datetime.fromisoformat(item.period_start.replace("Z", "+00:00"))), + utc_instant(datetime.fromisoformat(item.period_end.replace("Z", "+00:00"))), + ) + for item in owed + if item.stub + for buy_id in item.media_buy_ids + } + + candidates: list[_Candidate] = [] + for buy in selected: + listed = sorted({oid for _pkg, ids in buy.product_offering_ids() for oid in ids}) + wanted = [offering_id] if offering_id is not None else [o for o in listed if o in by_id] + if not wanted: + zone = account_timezone or "UTC" + if kind == "range": + periods = [value] + elif kind == "instant": + day = value.astimezone(require_zone(zone)).date() + periods = [_local_day(day, zone)] + else: + periods = [_local_day(value, zone)] + candidates.extend(_Candidate(buy, None, None, s, e, zone) for s, e in periods) + continue + for oid in wanted: + offering = by_id[oid] + reference = ( + value[0] + if kind == "range" + else ( + value if kind == "instant" else _local_day(value, account_timezone or "UTC")[0] + ) + ) + seen: set[tuple[datetime, datetime]] = set() + around = [reference, reference + timedelta(days=1)] + for zone in _zones(context, offering, around): + calendar = PeriodCalendar.for_schedule( + period_duration=offering.period_duration, + delivery_sla=offering.delivery_sla, + zone_name=zone, + anchor=offering.period_anchor + if offering.alignment == "billing_cycle" + else None, + local=offering.alignment != "utc", + ) + periods = _periods_for(calendar, kind, value, zone) + stub = stubs.get((buy.media_buy_id, oid)) + if kind == "range" and not periods and stub == tuple(value): + periods = [value] + for s, e in periods: + if (s, e) in seen: + continue + seen.add((s, e)) + candidates.append(_Candidate(buy, offering, calendar, s, e, zone)) + if kind == "range" and not seen: + raise CoverageInputError( + "invalid_input", + f"{iso_z(value[0])}/{iso_z(value[1])} is not a period of offering {oid!r}", + ) + + authorization = _authorization_projection(authorizations, status_retention_days) + + traces: list[CoverageTrace] = [] + for candidate in candidates: + key_offering = candidate.offering.offering_id if candidate.offering else "" + owed_item = owed_index.get( + ( + candidate.buy.media_buy_id, + key_offering, + candidate.start, + candidate.end, + candidate.zone, + ) + ) + elsewhere = None + if owed_item is None and candidate.offering is not None: + elsewhere = next( + ( + item + for (buy_id, oid, s, e, _zone), item in owed_index.items() + if buy_id == candidate.buy.media_buy_id + and oid == key_offering + and s < candidate.end + and e > candidate.start + ), + None, + ) + walk, version = _classify( + context, + candidate, + owed_item, + elsewhere, + now=now, + backfill=backfill_on_regrant, + invisible=candidate.buy.media_buy_id in hidden, + ) + assert walk.reason is not None + reason = _reason(walk.reason) + subject: dict[str, Any] = { + "account_id": account_id, + "subscription_id": context.subscription_id, + "subscription_version": version, + "media_buy_id": candidate.buy.media_buy_id, + "offering_id": candidate.offering.offering_id if candidate.offering else None, + "period": { + "start": iso_z(candidate.start), + "end": iso_z(candidate.end), + "timezone": candidate.zone, + }, + } + if principal_id is not None: + subject = {"principal_id": principal_id, **subject} + inputs = _inputs_document( + context, + candidate, + subject, + authorization=authorization, + backfill=backfill_on_regrant, + status_retention_days=status_retention_days, + ) + only = None + if side == "seller": + only = {"buy_visible": candidate.buy.media_buy_id not in hidden, **(seller_only or {})} + traces.append( + CoverageTrace( + side=side, + subject=subject, + verdict=reason.verdict, + reason=reason.code, + explanation=_explanation(reason.code, subject, walk.params), + steps=tuple(walk.steps), + inputs_digest=_digest(inputs), + seller_only=only, + ) + ) + traces.sort( + key=lambda t: ( + t.subject["period"]["start"], + str(t.subject["media_buy_id"]).encode(), + str(t.subject["offering_id"] or "").encode(), + t.subject["period"]["timezone"], + ) + ) + return traces + + +__all__ = [ + "EXPLAIN_TEMPLATES", + "EXPLAIN_TEMPLATES_VERSION", + "TRACE_VERSION", + "CoverageReason", + "CoverageSide", + "CoverageTrace", + "CoverageTraceDifference", + "CoverageTraceStep", + "CoverageVerdict", + "StepResult", + "coverage_reasons", + "diff_coverage_traces", + "explain_subscription_coverage", +] diff --git a/tests/test_reporting_subscription_explain.py b/tests/test_reporting_subscription_explain.py new file mode 100644 index 000000000..89cb7461c --- /dev/null +++ b/tests/test_reporting_subscription_explain.py @@ -0,0 +1,443 @@ +"""``CoverageTrace``: why a period is or is not owed, checked against the spec's vectors. + +For every canonical vector, every period of every offering a buy lists, near +the buy's flight, is explained, and the trace's verdict must equal the owed +rule's: owed exactly when the vector's expected rows owe it. The reason must +be a spec reason code whose verdict matches, taken from the first failing +rule in the spec's fixed order. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from typing import Any + +import pytest + +from adcp.reporting.subscriptions import COVERAGE_REASON_CODES, COVERAGE_RULE_ORDER +from adcp.reporting.subscriptions._clock import PeriodCalendar +from adcp.reporting.subscriptions.explain import ( + EXPLAIN_TEMPLATES, + CoverageTrace, + coverage_reasons, + diff_coverage_traces, + explain_subscription_coverage, +) +from tests.test_reporting_subscription_parity import ( + CASES, + SPEC, + at, + buys, + offerings, + pinned_tzdata, # noqa: F401 - the module fixture pins tzdata 2026f + windows, +) + +#: Periods examined per (buy, offering, zone) on each side of the owed rows. +_MARGIN = 3 +_CAP = 120 + + +def _kwargs(input_: dict[str, Any]) -> dict[str, Any]: + from adcp.reporting.subscriptions import SubscriptionAuthorization + + account = input_["account"] + return { + "subscription": windows(input_["versions"], input_.get("inactive_intervals", [])), + "account_id": account["account_id"], + "buys": buys(input_), + "offerings": offerings(input_), + "account_timezone": account.get("timezone"), + "account_operator": account.get("operator"), + "now": at(input_["now"]), + "authorizations": [ + SubscriptionAuthorization(at(grant["authorized_at"]), at(grant.get("unauthorized_at"))) + for grant in account.get("grants", ()) + ], + "coverage": input_.get("coverage"), + "backfill_on_regrant": bool(input_.get("backfill_on_regrant", False)), + "status_retention_days": input_.get("status_retention_days"), + } + + +def _calendars(kwargs: dict[str, Any], offering) -> list[PeriodCalendar]: + zones = {kwargs["account_timezone"]} + for entry in (kwargs.get("coverage") or {}).get("timezone_cutovers", ()): + zones.update({entry.get("timezone"), entry.get("previous_timezone")}) + overrides = {None} + for window in kwargs["subscription"]: + overrides.add((window.configuration.get("selector") or {}).get("period_timezone")) + result = [] + names = set() + for override in sorted(overrides, key=str): + for zone in sorted(z for z in zones if z) or [None]: + try: + calendar = offering.calendar(zone, override) + except Exception: # noqa: BLE001 - a zone this offering cannot use + continue + if calendar.zone_name not in names: + names.add(calendar.zone_name) + result.append(calendar) + return result + + +def _candidate_periods(case: dict[str, Any], kwargs: dict[str, Any]): + """(buy, offering, start, end) near each buy's flight and the owed rows.""" + by_id = {item.offering_id: item for item in kwargs["offerings"]} + now = kwargs["now"] + expected = case["expected"] + for buy in kwargs["buys"]: + listed = sorted( + {oid for _p, ids in buy.product_offering_ids() for oid in ids if oid in by_id} + ) + rows = [row for row in expected if row["media_buy_id"] == buy.media_buy_id] + anchors = [buy.start_time, min(buy.stop_time, now)] + anchors += [at(row["period_start"]) for row in rows] + anchors += [at(row["period_end"]) for row in rows] + low, high = min(anchors), max(anchors) + for row in rows: + if row.get("stub"): + # A billing stub is an owed period off the incoming calendar. + offering = by_id[row["offering_id"]] + yield buy, offering, at(row["period_start"]), at(row["period_end"]) + for oid in listed: + for calendar in _calendars(kwargs, by_id[oid]): + start = calendar.floor(low) + for _ in range(_MARGIN): + start = calendar.floor(start - timedelta(microseconds=1)) + seen = 0 + for period_start, period_end in calendar.periods( + starts_from=start, starts_before=high + timedelta(days=3) + ): + yield buy, by_id[oid], period_start, period_end + seen += 1 + if seen >= _CAP: + break + + +def _owed_rows(case: dict[str, Any]) -> set[tuple]: + return { + (row["media_buy_id"], row["offering_id"], at(row["period_start"]), at(row["period_end"])) + for row in case["expected"] + } + + +VECTOR_CASES = [case for case in CASES if "expected_error" not in case] + + +def test_the_vocabulary_is_the_spec_reason_codes_file() -> None: + document = json.loads((SPEC / "reason-codes.json").read_text()) + reasons = coverage_reasons() + assert [item["code"] for item in document["reasons"]] == list(reasons) + for item in document["reasons"]: + assert COVERAGE_REASON_CODES[item["code"]] == item.get("rule") + assert reasons[item["code"]].verdict == item["verdict"] + assert tuple(document["rule_order"]) == COVERAGE_RULE_ORDER + # Every reason renders one sentence. + assert set(EXPLAIN_TEMPLATES) == set(reasons) + + +@pytest.mark.parametrize("case", VECTOR_CASES, ids=[case["id"] for case in VECTOR_CASES]) +def test_trace_verdicts_match_the_owed_rule_on_every_vector(case: dict[str, Any]) -> None: + kwargs = _kwargs(case["input"]) + owed = _owed_rows(case) + explained: set[tuple] = set() + reasons = coverage_reasons() + for buy, offering, start, end in _candidate_periods(case, kwargs): + try: + traces = explain_subscription_coverage( + **kwargs, + period={"start": start, "end": end}, + media_buy_id=buy.media_buy_id, + offering_id=offering.offering_id, + ) + except Exception as error: # noqa: BLE001 + # A period of a zone the offering never uses, or outside any version. + if "not a period" in str(error): + continue + raise + for trace in traces: + key = (buy.media_buy_id, offering.offering_id, start, end) + is_owed = key in owed + assert trace.reason in reasons + assert reasons[trace.reason].verdict == trace.verdict + if trace.reason == "period_not_ended": + assert end > kwargs["now"] and not is_owed + continue + assert (trace.verdict == "owed") == is_owed, ( + case["id"], + trace.to_dict(), + case.get("rule"), + ) + # The steps follow the rule order, and only the last one fails. + rules = [step.rule for step in trace.steps] + assert rules == [rule for rule in COVERAGE_RULE_ORDER if rule in rules] + failed = [step for step in trace.steps if step.result == "fail"] + assert len(failed) == (0 if trace.verdict == "owed" else 1) + if failed: + assert failed[0] is trace.steps[-1] + rule = reasons[trace.reason].rule + assert rule == failed[0].rule or ( + trace.reason == "cutover_stretch" and failed[0].rule == "timezone" + ), (trace.reason, failed[0].rule) + if trace.verdict == "owed": + explained.add(key) + # Every owed row was examined and explained as owed. + assert owed <= explained, sorted(owed - explained) + + +def _scenario() -> dict[str, Any]: + from adcp.reporting.subscriptions import ( + SubscriptionBuy, + SubscriptionBuyPackage, + SubscriptionOffering, + ) + + offering = SubscriptionOffering( + offering_id="daily_pkg", + report_definition_id="rd_daily", + reporting_profile="standard", + feed_purpose="analytics", + period_duration="P1D", + alignment="account_timezone", + delivery_sla="PT6H", + supported_finality=frozenset({"snapshot", "official"}), + reconciliation_mode="delivery_only", + dimensions=frozenset({"package"}), + ) + state = { + "subscription_id": "analytics-daily", + "subscription_version": 1, + "state": "active", + "accepted_at": "2026-10-20T12:00:00Z", + "configuration": { + "subscription_id": "analytics-daily", + "subscription_version": 1, + "active": True, + "scope": {"all_authorized_accounts": True}, + "selector": { + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "snapshot", + "dimensions": ["package"], + }, + "reconciliation_mode": "delivery_only", + }, + } + buy = SubscriptionBuy( + media_buy_id="mb_123", + confirmed_at=datetime(2026, 11, 4, 10, tzinfo=timezone.utc), + start_time=datetime(2026, 11, 3, 5, tzinfo=timezone.utc), + end_time=datetime(2026, 11, 30, 5, tzinfo=timezone.utc), + packages=(SubscriptionBuyPackage("pkg_1", ("daily_pkg",)),), + ) + return { + "subscription": state, + "account_id": "acc_a", + "buys": [buy], + "offerings": [offering], + "account_timezone": "America/New_York", + "now": datetime(2026, 11, 10, tzinfo=timezone.utc), + } + + +def test_a_local_date_is_read_in_the_offering_zone_and_explained() -> None: + traces = explain_subscription_coverage(**_scenario(), period="2026-11-03") + assert len(traces) == 1 + trace = traces[0] + assert trace.subject["period"] == { + "start": "2026-11-03T05:00:00Z", + "end": "2026-11-04T05:00:00Z", + "timezone": "America/New_York", + } + assert (trace.verdict, trace.reason, trace.rule) == ( + "not_owed", + "before_confirmation", + "confirmation", + ) + assert trace.explanation == ( + "Not owed: the period ends 2026-11-04T05:00Z, before mb_123 was confirmed at " + "2026-11-04T10:00Z." + ) + assert [step.rule for step in trace.steps] == [ + "scope", + "authorization", + "version", + "activity", + "offering", + "flight", + "confirmation", + ] + assert trace.steps[-1].facts == {"confirmed_at": "2026-11-04T10:00:00Z"} + assert trace.inputs_digest.startswith("sha256:") + # The confirmation day itself is owed in full. + owed = explain_subscription_coverage(**_scenario(), period="2026-11-04")[0] + assert (owed.verdict, owed.reason) == ("owed", "owed") + assert owed.explanation.endswith("due at 2026-11-05T11:00Z.") + assert len(owed.steps) == len(COVERAGE_RULE_ORDER) + + +def test_a_period_that_has_not_ended_is_not_owed_yet() -> None: + trace = explain_subscription_coverage(**_scenario(), period="2026-11-10")[0] + assert (trace.verdict, trace.reason) == ("not_owed", "period_not_ended") + + +def test_buyer_and_seller_traces_of_one_period_agree_and_diff_names_the_first_rule() -> None: + buyer = explain_subscription_coverage(**_scenario(), period="2026-11-05")[0] + seller = explain_subscription_coverage( + **_scenario(), + period="2026-11-05", + side="seller", + principal_id="agency-x", + seller_only={"frozen_at": "2026-11-06T05:03:11Z"}, + )[0] + assert buyer.inputs_digest == seller.inputs_digest + assert diff_coverage_traces(buyer, seller) is None + assert seller.seller_only == {"buy_visible": True, "frozen_at": "2026-11-06T05:03:11Z"} + assert "seller_only" not in seller.for_buyer().to_dict() + # A stale buyer read (the buy confirmed later) differs at the confirmation rule. + stale = _scenario() + from dataclasses import replace + + stale["buys"] = [ + replace(stale["buys"][0], confirmed_at=datetime(2026, 11, 6, 10, tzinfo=timezone.utc)) + ] + late = explain_subscription_coverage(**stale, period="2026-11-05")[0] + difference = diff_coverage_traces(late, seller) + assert difference is not None + assert difference.rule == "confirmation" + assert not difference.same_inputs + assert "different state" in difference.describe() + + +def test_a_trace_round_trips_through_json() -> None: + trace = explain_subscription_coverage(**_scenario(), period="2026-11-03")[0] + again = CoverageTrace.from_dict(json.loads(trace.to_json())) + assert again == trace + + +def test_seller_only_reason_for_a_buy_the_caller_cannot_see() -> None: + trace = explain_subscription_coverage( + **_scenario(), period="2026-11-05", side="seller", invisible_media_buy_ids={"mb_123"} + )[0] + assert (trace.reason, trace.rule) == ("buy_not_visible", "scope") + with pytest.raises(ValueError, match="seller-only"): + explain_subscription_coverage( + **_scenario(), period="2026-11-05", invisible_media_buy_ids={"mb_123"} + ) + + +def test_a_range_must_be_a_period_of_the_schedule() -> None: + from adcp.reporting.subscriptions import CoverageInputError + + with pytest.raises(CoverageInputError, match="not a period"): + explain_subscription_coverage( + **_scenario(), period={"start": "2026-11-03T00:00:00Z", "end": "2026-11-04T00:00:00Z"} + ) + + +def test_the_templates_are_one_sentence_each() -> None: + for code, template in EXPLAIN_TEMPLATES.items(): + assert template.endswith("."), code + + +@pytest.mark.asyncio +async def test_the_managed_seller_explains_what_its_ledger_owes() -> None: + from adcp.reporting import PrincipalRef + from adcp.reporting.subscriptions import SubscriptionBuy + from tests.test_reporting_subscription_managed import ACCOUNT, AGENCY, START, _ctx, _World + + world = _World() + await world.subscribe() + world.clock.set(START + timedelta(hours=3, minutes=30)) + turn = await world.service.run_worker() + assert turn.subscription_obligations_written == 2 + principal = PrincipalRef(None, AGENCY) + status = await world.platform.get_reporting_status({"view": "periods"}, _ctx()) + ledger = {item["reporting_obligation_id"] for item in status["periods"]} + + traces = await world.service.explain_coverage(principal, ACCOUNT, period="2026-11-01") + owed = [trace for trace in traces if trace.verdict == "owed"] + assert {trace.seller_only["reporting_obligation_id"] for trace in owed} == ledger + assert all(trace.side == "seller" and trace.seller_only["buy_visible"] for trace in traces) + # The first hour ends before the buy starts; the fourth has not ended. + reasons = {trace.subject["period"]["start"]: trace.reason for trace in traces} + assert reasons["2026-11-01T00:00:00Z"] == "outside_flight" + assert reasons["2026-11-01T03:00:00Z"] == "after_stop" + one = await world.service.explain_coverage( + principal, ACCOUNT, period="2026-11-01T01:30:00Z", subscription_id="analytics-hourly" + ) + assert [(t.verdict, t.reason) for t in one] == [("owed", "owed")] + assert "seller_only" not in one[0].for_buyer().to_dict() + # A buy the caller cannot see is explained, seller-side only. + hidden = SubscriptionBuy( + media_buy_id="media-buy-ui", + confirmed_at=START, + start_time=START, + end_time=START + timedelta(hours=3), + ) + unseen = await world.service.explain_coverage( + principal, + ACCOUNT, + period="2026-11-01T01:30:00Z", + media_buy_id="media-buy-ui", + invisible_buys=[hidden], + ) + assert [t.reason for t in unseen] == ["buy_not_visible"] + with pytest.raises(KeyError): + await world.service.explain_coverage( + principal, ACCOUNT, period="2026-11-01", subscription_id="missing" + ) + + +def test_the_cli_diff_names_the_first_differing_rule(tmp_path, capsys) -> None: + from adcp.reporting.cli import main + + buyer = explain_subscription_coverage(**_scenario(), period="2026-11-05")[0] + seller = explain_subscription_coverage( + **_scenario(), period="2026-11-05", side="seller", principal_id="agency-x" + )[0] + (tmp_path / "buyer.json").write_text(buyer.to_json()) + (tmp_path / "seller.json").write_text(json.dumps([seller.to_dict()])) + assert main(["diff", str(tmp_path / "buyer.json"), str(tmp_path / "seller.json")]) == 0 + assert "agree" in capsys.readouterr().out + stale = _scenario() + from dataclasses import replace + + stale["buys"] = [ + replace(stale["buys"][0], confirmed_at=datetime(2026, 11, 6, 10, tzinfo=timezone.utc)) + ] + late = explain_subscription_coverage(**stale, period="2026-11-05")[0] + (tmp_path / "late.json").write_text(late.to_json()) + assert main(["diff", str(tmp_path / "late.json"), str(tmp_path / "seller.json")]) == 1 + out = capsys.readouterr().out + assert "rule 'confirmation'" in out and "different state" in out + + +def test_the_cli_explains_from_the_seller_service(monkeypatch, capsys) -> None: + import sys + import types + + from adcp.reporting.cli import main + + calls: list[tuple] = [] + + class Service: + async def explain_coverage(self, principal, account_id, **kwargs): + calls.append((principal, account_id, kwargs)) + return explain_subscription_coverage( + **_scenario(), period=kwargs["period"], side="seller", principal_id="agency-x" + ) + + module = types.ModuleType("explain_cli_app") + module.service = Service() + monkeypatch.setitem(sys.modules, "explain_cli_app", module) + argv = ["explain", "--app", "explain_cli_app:service", "--principal", "agency-x"] + assert main([*argv, "--account", "acc_a", "--period", "2026-11-03"]) == 0 + out = capsys.readouterr().out + assert "before_confirmation" in out and "was confirmed at 2026-11-04T10:00Z" in out + assert calls[0][1] == "acc_a" and calls[0][0].principal_id == "agency-x" + assert main([*argv, "--account", "acc_a", "--period", "2026-11-03", "--json"]) == 0 + document = json.loads(capsys.readouterr().out) + assert document[0]["reason"] == "before_confirmation" From 90c16ec82709749ef85f510784eef421136fb666 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 17:08:08 +0000 Subject: [PATCH 2/6] test(reporting): run the row-upload lease test at the wall clock The producer row-upload test stamps its worker lease with the settling harness clock, fixed at 2026-11-01T02:10Z, while the row store checks the lease against the database's clock_timestamp. Once the wall clock passed that instant plus the lease, every run failed with STATE_UNAVAILABLE: no live worker lease covers this row upload, on next and on every pull request. The test now runs its turn at the wall clock, so the lease lies in the real future. Co-Authored-By: Claude Opus 5.5 (1M context) (cherry picked from commit ae15bab6e8a73f974ff6adc30cf761454b089b44) --- .../reporting/test_reporting_row_storage_object_pg.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/tests/conformance/reporting/test_reporting_row_storage_object_pg.py b/tests/conformance/reporting/test_reporting_row_storage_object_pg.py index aa2410045..f4cab4017 100644 --- a/tests/conformance/reporting/test_reporting_row_storage_object_pg.py +++ b/tests/conformance/reporting/test_reporting_row_storage_object_pg.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -442,13 +443,18 @@ async def test_the_producer_stages_object_uploads_outside_its_publication_transa from tests.test_reporting_settling import _capabilities, _harness, _revisions async with isolated_reporting_pool() as pool: - producer, store, _, _ = await _harness( + producer, store, _, clock = await _harness( _capabilities(restatement_window="P3D"), store_factory=lambda clock: PgReportingLedgerStore( pool=pool, clock=clock, row_storage=options(tmp_path) ), ) assert isinstance(producer, ReportingProducer) + # The row upload checks the worker lease against the database's + # clock_timestamp(), so the lease the injected clock stamps must lie in + # the real future: run the turn at the wall clock, not the harness's + # fixed instant, which is already in the past. + clock[0] = datetime.now(timezone.utc) turn = await producer.run_worker() assert turn.revisions_committed (revision,) = await _revisions(store) From f1c23ca5e9237c34a12917c4296aed9ac6eb1c23 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 17:22:17 +0000 Subject: [PATCH 3/6] fix(reporting): fold whether the period has ended into the trace digest A period the owed rule owes is period_not_ended before its end and owed after it, so two traces taken on either side of the end carried equal inputs_digest but different verdicts, and diff_coverage_traces blamed an SDK logic bug for clock skew. The digest now includes whether the period has ended; now itself still is not part of it. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reporting-subscriptions-coverage.md | 2 +- src/adcp/reporting/subscriptions/explain.py | 11 +++++++++-- tests/test_reporting_subscription_explain.py | 15 +++++++++++++++ 3 files changed, 25 insertions(+), 3 deletions(-) diff --git a/docs/reporting-subscriptions-coverage.md b/docs/reporting-subscriptions-coverage.md index e2018fc35..2e029b7f6 100644 --- a/docs/reporting-subscriptions-coverage.md +++ b/docs/reporting-subscriptions-coverage.md @@ -69,7 +69,7 @@ python -m adcp reporting explain --app myapp.reporting:service \ - Steps stop at the first failing rule. An owed period passes every rule; `timezone` is `not_applicable` unless the offering is `account_timezone` aligned. - `seller_only` holds whether the caller can see the buy and, for an owed period, its `reporting_obligation_id`. It never leaves the seller: send `trace.for_buyer()`. -- `inputs_digest` hashes only wire-visible facts local to the period: the version in effect, the authorization as `list_accounts` and `accounts_unauthorized` show it, the zone in effect, the buy as `get_media_buys` returns it, the offering, the `excluded[]` entries covering the period start, and whether an authorization gap covers it. `now` is not part of it. +- `inputs_digest` hashes only wire-visible facts local to the period: the version in effect, the authorization as `list_accounts` and `accounts_unauthorized` show it, the zone in effect, the buy as `get_media_buys` returns it, the offering, the `excluded[]` entries covering the period start, and whether an authorization gap covers it. `now` is not part of it, but whether the period has ended is, so two traces taken on either side of the period end never share a digest. ### Comparing a buyer trace with a seller trace diff --git a/src/adcp/reporting/subscriptions/explain.py b/src/adcp/reporting/subscriptions/explain.py index 90df9760a..03ef5d762 100644 --- a/src/adcp/reporting/subscriptions/explain.py +++ b/src/adcp/reporting/subscriptions/explain.py @@ -16,7 +16,8 @@ ``reason-codes.json``, packaged with the SDK for the AdCP release it serves. Explanations are rendered from :data:`EXPLAIN_TEMPLATES`, one sentence per reason, so both sides emit identical text for identical facts. ``now`` is not -part of the digest: only ``period_not_ended`` depends on it. +part of the digest, but whether the period has ended is, so +``period_not_ended`` is never a same-inputs difference. ``seller_only`` facts (when the seller froze a period, whether a buy is visible to the caller, the obligation id) never enter the digest, and a @@ -1023,6 +1024,7 @@ def _inputs_document( authorization: Mapping[str, Any], backfill: bool, status_retention_days: int | None, + ended: bool, ) -> dict[str, Any]: """The wire-visible facts one period's classification depends on. @@ -1032,7 +1034,10 @@ def _inputs_document( the zone in effect, the buy as ``get_media_buys`` returns it, the offering, and the ``excluded[]`` entries covering the period start. An authorization gap enters as whether it covers the period, which the - seller's ``gaps[]`` and its grant journal state identically. + seller's ``gaps[]`` and its grant journal state identically. ``now`` + itself is not part of it, but whether the period has ended is: two traces + taken on either side of the period end differ in their inputs, never only + in their verdict. """ start, end = candidate.start, candidate.end window = _window_at(context.windows, start) @@ -1094,6 +1099,7 @@ def _inputs_document( "authorization_gap": gap, "backfill_on_regrant": backfill, "status_retention_days": status_retention_days, + "period_ended": ended, } @@ -1334,6 +1340,7 @@ def explain_subscription_coverage( authorization=authorization, backfill=backfill_on_regrant, status_retention_days=status_retention_days, + ended=candidate.end <= now, ) only = None if side == "seller": diff --git a/tests/test_reporting_subscription_explain.py b/tests/test_reporting_subscription_explain.py index 89cb7461c..2a79e949d 100644 --- a/tests/test_reporting_subscription_explain.py +++ b/tests/test_reporting_subscription_explain.py @@ -441,3 +441,18 @@ async def explain_coverage(self, principal, account_id, **kwargs): assert main([*argv, "--account", "acc_a", "--period", "2026-11-03", "--json"]) == 0 document = json.loads(capsys.readouterr().out) assert document[0]["reason"] == "before_confirmation" + + +def test_traces_on_either_side_of_the_period_end_differ_in_their_inputs() -> None: + before = _scenario() | {"now": datetime(2026, 11, 5, 6, tzinfo=timezone.utc)} + after = _scenario() | {"now": datetime(2026, 11, 6, 6, tzinfo=timezone.utc)} + open_ = explain_subscription_coverage(**before, period="2026-11-05")[0] + ended = explain_subscription_coverage(**after, period="2026-11-05")[0] + assert (open_.reason, ended.reason) == ("period_not_ended", "owed") + difference = diff_coverage_traces(open_, ended) + assert difference is not None and not difference.same_inputs + # Later, with the period still ended, the digest is unchanged. + later = _scenario() | {"now": datetime(2026, 11, 20, tzinfo=timezone.utc)} + assert explain_subscription_coverage(**later, period="2026-11-05")[0].inputs_digest == ( + ended.inputs_digest + ) From 9eb96a012f48454c3c825e83919bcf5047dd1fde Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 18:25:58 +0000 Subject: [PATCH 4/6] fix(reporting): explain periods before an open revocation as owed:withheld reason-codes.json defines owed:withheld as an owed period that ended at or before the latest unauthorized_at with no re-grant since. explain used it only for periods not yet due at the loss, and called the rest owed. It now follows the definition, which the TypeScript explainer also applies. The trace digest's offering facts carry only the zone the offering publishes, so an account-resolved source zone no longer makes the two SDKs' digests differ. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/explain.py | 13 +++++++++++-- tests/test_reporting_subscription_explain.py | 15 +++++++++++++++ 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/src/adcp/reporting/subscriptions/explain.py b/src/adcp/reporting/subscriptions/explain.py index 03ef5d762..e19024c96 100644 --- a/src/adcp/reporting/subscriptions/explain.py +++ b/src/adcp/reporting/subscriptions/explain.py @@ -475,7 +475,14 @@ def _offering_facts(offering: SubscriptionOffering | None) -> dict[str, Any] | N "period_duration": offering.period_duration, "alignment": offering.alignment, "delivery_sla": offering.delivery_sla, - "period_timezone": offering.period_timezone, + # The zone the offering publishes; an account-resolved source zone is + # not on the wire, so it is not part of the facts. + "period_timezone": ( + None + if offering.alignment == "source_timezone" + and offering.period_timezone_policy != "fixed" + else offering.period_timezone + ), "period_anchor": _z(offering.period_anchor), "available_from": _z(offering.available_from), "withdrawn_at": _z(offering.withdrawn_at), @@ -972,7 +979,9 @@ def _classify( walk.params.update(expected_at=owed.expected_at) if end > now: walk.reason = "period_not_ended" - elif owed.withheld: + elif owed.withheld or (chains and chains[-1].end is not None and end <= chains[-1].end): + # reason-codes.json: the period ended at or before the latest + # unauthorized_at and no re-grant has followed yet. last = chains[-1] if chains else None walk.reason = "owed:withheld" walk.params.update(unauthorized_at=None if last is None else last.end) diff --git a/tests/test_reporting_subscription_explain.py b/tests/test_reporting_subscription_explain.py index 2a79e949d..3ae0b79b3 100644 --- a/tests/test_reporting_subscription_explain.py +++ b/tests/test_reporting_subscription_explain.py @@ -456,3 +456,18 @@ def test_traces_on_either_side_of_the_period_end_differ_in_their_inputs() -> Non assert explain_subscription_coverage(**later, period="2026-11-05")[0].inputs_digest == ( ended.inputs_digest ) + + +def test_a_period_ended_before_an_open_revocation_is_owed_withheld() -> None: + from adcp.reporting.subscriptions import SubscriptionAuthorization + + scenario = _scenario() + scenario["authorizations"] = [ + SubscriptionAuthorization( + datetime(2026, 10, 1, tzinfo=timezone.utc), datetime(2026, 11, 7, tzinfo=timezone.utc) + ) + ] + scenario["status_retention_days"] = 30 + trace = explain_subscription_coverage(**scenario, period="2026-11-05")[0] + assert (trace.verdict, trace.reason) == ("owed", "owed:withheld") + assert "lost authority for acc_a at 2026-11-07T00:00Z" in trace.explanation From 0974a2d5bfe20a31c9227e845c70ae9f79adeca7 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 18:38:58 +0000 Subject: [PATCH 5/6] test(reporting): pin the account-resolved source zone out of the trace digest Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/test_reporting_subscription_explain.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/test_reporting_subscription_explain.py b/tests/test_reporting_subscription_explain.py index 3ae0b79b3..b265f07f1 100644 --- a/tests/test_reporting_subscription_explain.py +++ b/tests/test_reporting_subscription_explain.py @@ -471,3 +471,23 @@ def test_a_period_ended_before_an_open_revocation_is_owed_withheld() -> None: trace = explain_subscription_coverage(**scenario, period="2026-11-05")[0] assert (trace.verdict, trace.reason) == ("owed", "owed:withheld") assert "lost authority for acc_a at 2026-11-07T00:00Z" in trace.explanation + + +def test_an_account_resolved_source_zone_stays_out_of_the_digest() -> None: + from dataclasses import replace + + from adcp.reporting.subscriptions.explain import _offering_facts + + base = _scenario()["offerings"][0] + resolved = replace( + base, + alignment="source_timezone", + period_timezone="America/New_York", + period_timezone_policy="account_resolved", + ) + other = replace(resolved, period_timezone="Europe/Paris") + # The seller resolved the zone; a buyer may not have: the facts agree. + assert _offering_facts(resolved) == _offering_facts(other) + assert _offering_facts(resolved)["period_timezone"] is None + fixed = replace(resolved, period_timezone_policy="fixed") + assert _offering_facts(fixed)["period_timezone"] == "America/New_York" From fc370c61a7e55f082ac67011fbb2d571828e110a Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 18:51:31 +0000 Subject: [PATCH 6/6] fix(reporting): refuse coverage input only where a schedule reads it Differential testing in adcp#8679 found three coverage input errors raised for values no schedule reads. The account timezone and each timezone cutover are now validated only where an account_timezone calendar is built, and previous_timezone is required only for a cutover that applies to such an offering. An offering without standard_dimensions constructs, and is refused with missing_standard_dimensions only when it is a candidate for the selector, as error-codes.json states. Each minimized repro from adcp-client-python#1769 is a test. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/coverage.py | 50 +++++++++++++------ .../reporting/subscriptions/resolution.py | 28 ++++++++--- tests/test_reporting_subscription_coverage.py | 15 +++++- tests/test_reporting_subscription_parity.py | 41 +++++++++++++++ 4 files changed, 111 insertions(+), 23 deletions(-) diff --git a/src/adcp/reporting/subscriptions/coverage.py b/src/adcp/reporting/subscriptions/coverage.py index 37b9086dd..a313e5565 100644 --- a/src/adcp/reporting/subscriptions/coverage.py +++ b/src/adcp/reporting/subscriptions/coverage.py @@ -872,6 +872,36 @@ def _timezone_at( return zone +def _offering_zone( + item: SubscriptionOffering, + instant: datetime, + account_timezone: str | None, + changes: Sequence[AccountTimezoneChange], + period_timezone: str | None, +) -> str | None: + """The account zone an offering's calendar reads at ``instant``, validated. + + Only an ``account_timezone`` offering without a ``period_timezone`` + override reads it; for any other the zone is irrelevant and never checked. + """ + if item.alignment != "account_timezone" or period_timezone is not None: + return account_timezone + applicable = [change for change in changes if change.offering_id in {None, item.offering_id}] + for change in applicable: + if change.previous_timezone is None: + raise CoverageInputError( + "missing_previous_timezone", + "a timezone change needs previous_timezone: the account's current timezone " + "already shows the new zone", + ) + require_zone(change.previous_timezone, field="timezone_cutovers.previous_timezone") + require_zone(change.timezone, field="timezone_cutovers.timezone") + zone = _timezone_at(instant, account_timezone, changes, item.offering_id) + if zone is not None and not applicable: + require_zone(zone, field="account_timezone") + return zone + + def _windows( subscription: Any, version_configurations: Mapping[int, Any] | None ) -> tuple[SubscriptionVersionWindow, ...]: @@ -967,19 +997,9 @@ def _coverage_context( previous_timezone=None if previous is None else _text(previous), ) ) - changes.sort(key=lambda item: item.changed_at) - if account_timezone is not None: - require_zone(account_timezone, field="account_timezone") - for change in changes: - require_zone(change.timezone, field="timezone_cutovers.timezone") - for change in changes: - if change.previous_timezone is None: - raise CoverageInputError( - "missing_previous_timezone", - "a timezone change needs previous_timezone: the account's current timezone " - "already shows the new zone", - ) - require_zone(change.previous_timezone, field="timezone_cutovers.previous_timezone") + # error-codes.json: a zone is refused only when a schedule reads it, so the + # account zone and each cutover are validated where an account_timezone + # calendar is built (_offering_zone), never up front. changes.sort(key=lambda item: item.changed_at) every_removal = subscription_exceptions(excluded, gap_entries) removals = tuple(item for item in every_removal if not item.package_ids) @@ -1308,7 +1328,9 @@ def _regimes( _Resolved( item, item.calendar( - _timezone_at(start, account_timezone, changes, item.offering_id), + _offering_zone( + item, start, account_timezone, changes, selector.period_timezone + ), selector.period_timezone, ), ) diff --git a/src/adcp/reporting/subscriptions/resolution.py b/src/adcp/reporting/subscriptions/resolution.py index 3b294d0dc..1e6cc88f0 100644 --- a/src/adcp/reporting/subscriptions/resolution.py +++ b/src/adcp/reporting/subscriptions/resolution.py @@ -125,6 +125,10 @@ class SubscriptionOffering: method_transport: str | None = None available_from: datetime | None = None withdrawn_at: datetime | None = None + #: Whether the offering published ``standard_dimensions``. One that did + #: not is refused (``missing_standard_dimensions``) only when it is a + #: candidate for a selector. + dimensions_published: bool = True def __post_init__(self) -> None: if not self.offering_id: @@ -195,16 +199,15 @@ def from_offering( offering = offering.model_dump(mode="json", exclude_none=True) if not isinstance(offering, Mapping): raise TypeError("offering must be a ReportingDeliveryOffering or a mapping") + published = True if dimensions is None: if "standard_dimensions" not in offering: # adcp#8585: sellers advertising subscriptions MUST publish it, - # and matching never guesses a breakdown. - raise CoverageInputError( - "missing_standard_dimensions", - f"offering {offering.get('offering_id')!r} has no standard_dimensions. " - "The seller must publish standard_dimensions on every offering.", - ) - dimensions = tuple(offering["standard_dimensions"] or ()) + # and matching never guesses a breakdown. error-codes.json + # refuses only a *candidate* offering without it, so the + # refusal waits until the offering is one. + published = False + dimensions = tuple(offering.get("standard_dimensions") or ()) schedule = offering["schedule"] fixed_zone = schedule.get("period_timezone") if fixed_zone is not None and period_timezone is not None and fixed_zone != period_timezone: @@ -233,6 +236,7 @@ def from_offering( datetime.fromisoformat(str(anchor).replace("Z", "+00:00")) if anchor else None ), period_anchor_policy=schedule.get("period_anchor_policy"), + dimensions_published=published, period_timezone_policy=schedule.get("period_timezone_policy"), method_pattern=method.get("pattern"), method_transport=method.get("transport"), @@ -477,7 +481,15 @@ def _is_candidate( or offering.method_transport != destination.transport ): return False - return offering.can_produce(selector.period_timezone) + if not offering.can_produce(selector.period_timezone): + return False + if not offering.dimensions_published: + raise CoverageInputError( + "missing_standard_dimensions", + f"offering {offering.offering_id!r} has no standard_dimensions. " + "The seller must publish standard_dimensions on every offering.", + ) + return True def resolve_subscription_offerings( diff --git a/tests/test_reporting_subscription_coverage.py b/tests/test_reporting_subscription_coverage.py index 40df653e4..42d305d39 100644 --- a/tests/test_reporting_subscription_coverage.py +++ b/tests/test_reporting_subscription_coverage.py @@ -1499,8 +1499,21 @@ def test_wire_offering_without_standard_dimensions_is_refused() -> None: "supported_finality": ["official"], "reconciliation_mode": "delivery_only", } + from adcp.reporting.subscriptions import resolve_subscription_offerings + + # error-codes.json refuses only a candidate offering without it (adcp#8679). + legacy = SubscriptionOffering.from_offering(wire) + assert not legacy.dimensions_published + selector = { + "selector": { + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + }, + "reconciliation_mode": "delivery_only", + } with pytest.raises(ValueError, match="standard_dimensions"): - SubscriptionOffering.from_offering(wire) + resolve_subscription_offerings(selector, [legacy], product_offering_ids=["legacy"]) assert SubscriptionOffering.from_offering(wire, dimensions=()).dimensions == frozenset() diff --git a/tests/test_reporting_subscription_parity.py b/tests/test_reporting_subscription_parity.py index 53ec5dcdb..ffbfe8e71 100644 --- a/tests/test_reporting_subscription_parity.py +++ b/tests/test_reporting_subscription_parity.py @@ -378,3 +378,44 @@ def test_seller_configuration_codes_match_the_frozen_spec_vocabulary() -> None: assert (error.code, error.recovery, error.dispatched) == (code, recovery, None) with pytest.raises(ValueError): ReliableReportingConfigurationError("x", code="NOT_A_CODE") + + +def _baseline() -> dict[str, Any]: + import copy + + return copy.deepcopy(next(case for case in CASES if case["id"] == "baseline")["input"]) + + +def _starts(input_: dict[str, Any]) -> list[str]: + return sorted({p.period_start for p in periods(input_)}) + + +BASELINE_STARTS = ["2026-11-03T00:00:00Z", "2026-11-04T00:00:00Z", "2026-11-05T00:00:00Z"] + + +def test_issue_1769_an_account_zone_no_schedule_reads_is_not_validated() -> None: + input_ = _baseline() + input_["account"]["timezone"] = "+05:00" + assert _starts(input_) == BASELINE_STARTS + + +def test_issue_1769_a_cutover_no_account_timezone_offering_reads_needs_no_previous_zone() -> None: + input_ = _baseline() + input_["coverage"] = { + "timezone_cutovers": [ + {"effective_at": "2026-11-05T05:00:00Z", "timezone": "America/New_York"} + ] + } + assert _starts(input_) == BASELINE_STARTS + + +def test_issue_1769_a_non_candidate_offering_without_standard_dimensions_is_ignored() -> None: + import copy + + input_ = _baseline() + extra = copy.deepcopy(input_["offerings"][0]) + extra.update(offering_id="off_z_pacing_hourly", feed_purpose="pacing") + extra["schedule"] = {**extra["schedule"], "period_duration": "PT1H"} + extra.pop("standard_dimensions", None) + input_["offerings"].append(extra) + assert _starts(input_) == BASELINE_STARTS