diff --git a/docs/reporting-subscriptions-coverage.md b/docs/reporting-subscriptions-coverage.md index 242dfe4e4..2e029b7f6 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, 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 + +``` +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..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, ...]: @@ -888,72 +918,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) @@ -974,19 +997,9 @@ def owed_subscription_periods( 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) @@ -1033,25 +1046,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): @@ -1224,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/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..e19024c96 --- /dev/null +++ b/src/adcp/reporting/subscriptions/explain.py @@ -0,0 +1,1394 @@ +"""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, 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 +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, + # 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), + } + + +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 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) + 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, + ended: bool, +) -> 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. ``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) + 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, + "period_ended": ended, + } + + +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, + ended=candidate.end <= now, + ) + 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/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/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) 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_explain.py b/tests/test_reporting_subscription_explain.py new file mode 100644 index 000000000..b265f07f1 --- /dev/null +++ b/tests/test_reporting_subscription_explain.py @@ -0,0 +1,493 @@ +"""``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" + + +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 + ) + + +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 + + +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" 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