From dbf329f11a5d603ca795d075bfed3ce23bf9312e Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 17:19:37 +0000 Subject: [PATCH 1/5] feat(reporting): add buyer reconcile and explain on client.reporting client.reporting.reconcile reports every owed period of every covered account with a status, the spec reason code, the explanation sentence and the responsible party. Everything is read from the wire on each run: the roster is list_accounts intersected with the subscription scopes plus their accounts_unauthorized entries, grant history is authorized_at, regranted_at and unauthorized_at, and buys, offerings, coverage and obligations come from get_media_buys, capabilities and a complete get_reporting_status walk. The owed set is the shared owed rule; lines are delivered, pending, late, missing_obligation or withheld, and not_owed with its reason on request. grace defaults to one hour. definitive is true only when every read completed, and issues name what to retry. An account whose authority was lost reads from the last facts cached for it, with its periods withheld. client.reporting.explain returns the buyer CoverageTrace, identical to the seller trace for the same facts. CoverageExplainer prepares one coverage input for many classifications. ReportingSubscriptionTracker becomes an internal detail. CLI: python -m adcp reporting reconcile and explain with --agent. Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 5 +- docs/reporting-subscriptions-buyer.md | 42 +- src/adcp/reporting/cli.py | 73 +- src/adcp/reporting/subscriptions/__init__.py | 14 + src/adcp/reporting/subscriptions/consumer.py | 7 +- src/adcp/reporting/subscriptions/explain.py | 497 ++++++---- src/adcp/reporting/subscriptions/facade.py | 91 +- src/adcp/reporting/subscriptions/reconcile.py | 899 ++++++++++++++++++ tests/reporting_inprocess.py | 58 ++ .../test_reporting_subscription_reconcile.py | 196 ++++ 10 files changed, 1670 insertions(+), 212 deletions(-) create mode 100644 src/adcp/reporting/subscriptions/reconcile.py create mode 100644 tests/reporting_inprocess.py create mode 100644 tests/test_reporting_subscription_reconcile.py diff --git a/README.md b/README.md index e15a48e57..df1e8abdc 100644 --- a/README.md +++ b/README.md @@ -365,12 +365,15 @@ platform = reporting.install(MySeller()) # serves get_principal/sync_principal, ```python await client.reporting.subscribe(analytics_daily, register_destination=archive) await client.reporting.unsubscribe("pacing-hourly") +report = await client.reporting.reconcile() # every owed period: delivered, late, missing, withheld +traces = await client.reporting.explain(account_id="acc_a", period="2026-11-03") # why, rule by rule outcome = await client.accounts.ensure(key, billing="operator") # AccountReady | AccountPending | ... ``` - **Storage.** One object; process-local storage is refused in production. Apply schema changes with `python -m adcp migrate`. - **Grants.** The SDK is the only source of `authorized_at`. End a relationship with `wind_down_grant()` instead of revoking. -- **Coming in later betas:** buyer `reconcile`, `explain` and events, the test kit, and explain traces. +- **Explain.** Seller `reporting.explain_coverage(...)` and buyer `client.reporting.explain(...)` return the same `CoverageTrace`; `python -m adcp reporting diff` compares them. +- **Coming in later betas:** buyer events and the test kit. Guides: [seller](docs/reporting-subscriptions-seller.md) · [buyer](docs/reporting-subscriptions-buyer.md) · [coverage](docs/reporting-subscriptions-coverage.md) · [accounts](docs/accounts-ensure-and-change.md) · [migrating from account-level configs](docs/reporting-subscriptions-migration.md) · [testing](docs/reporting-subscriptions-testing.md) · [errors](docs/reporting-subscriptions-errors.md) diff --git a/docs/reporting-subscriptions-buyer.md b/docs/reporting-subscriptions-buyer.md index 53b990f6b..886e1a260 100644 --- a/docs/reporting-subscriptions-buyer.md +++ b/docs/reporting-subscriptions-buyer.md @@ -49,7 +49,47 @@ An `all_authorized_accounts` subscription follows every grant, including account ## Reconcile -`client.reporting.reconcile()` and `explain()` arrive in a later beta. Until then, `owed_subscription_periods(...)` and `ReportingSubscriptionTracker` compute owed periods from your own records (see [Coverage, explained](reporting-subscriptions-coverage.md#the-owed-rules)). +```python +report = await client.reporting.reconcile(since="2026-11-01") +print(report.definitive, report.summary) +for line in report.periods: + if line.responsible_party == "seller": + print(line.account_id, line.period["start"], line.status, line.reason, line.message) +``` + +`reconcile()` reads everything from the wire on each run. Never compute periods yourself. +- **Roster:** `list_accounts` ∩ your subscriptions' scopes, plus each subscription's `accounts_unauthorized[]`. There is no account list to keep. +- **Grant history:** `authorization.authorized_at` and `regranted_at` from `list_accounts`, and `unauthorized_at` from `accounts_unauthorized[]`. You never record grants. +- **Facts:** buys from `get_media_buys`, offerings from capabilities, and `subscription_coverage[]` and the obligations from a complete `get_reporting_status` periods walk. + +Each owed period that has ended becomes a `PeriodLine`, with wire-style keys and `to_dict()`: + +| `status` | Meaning | `responsible_party` | +|---|---|---| +| `delivered` | a current revision meets the selector's `required_finality` | `none`, or `buyer` when a billing receipt is owed (`receipt.status: owed_by_buyer`) | +| `pending` | not yet due: within `grace` of the period end, or of `expected_at` | `none` | +| `late` | an obligation exists, but its revision is more than `grace` past `expected_at` | `seller` | +| `missing_obligation` | no obligation `grace` after the period ended | `seller` | +| `withheld` | owed, and your authority is currently lost | `none` | +| `not_owed` | only with `include_not_owed=True`: every other period of each buy's flight, with its reason | `none` | + +- `reason` is the spec reason code and `message` its sentence, the same as `explain`. +- `grace` defaults to one hour, never zero, so a fresh period is not flagged as missing. +- `report.definitive` is true only when the roster, every account's obligations walk and every owed set completed. Otherwise `report.issues` say what to retry; a partial roster is never used. +- An account whose authority you lost is reported from the last facts read for it, with its unpaid periods `withheld`. + +From a shell: `python -m adcp reporting reconcile --agent [--since 2026-11-01] [--json]`. + +## Explain + +```python +traces = await client.reporting.explain(account_id="acc_a", subscription_id="analytics-daily", + media_buy_id="mb_1", period="2026-11-03") +``` + +`explain()` returns the same `CoverageTrace` the seller's `explain_coverage` builds, with `side="buyer"`. Compare the two with `diff_coverage_traces` or `python -m adcp reporting diff buyer.json seller.json`. See [Coverage, explained](reporting-subscriptions-coverage.md#explaining-a-period). The CLI form is `python -m adcp reporting explain --agent --account --period `. + +`ReportingSubscriptionTracker` is now an internal detail of these two calls. It stays importable for code that already uses it. ## Events diff --git a/src/adcp/reporting/cli.py b/src/adcp/reporting/cli.py index 402e0e859..a17a6b9c4 100644 --- a/src/adcp/reporting/cli.py +++ b/src/adcp/reporting/cli.py @@ -10,6 +10,11 @@ a coroutine function), the way ASGI servers name an app. ``--tenant`` sets the principal's tenant. +Buyer side, against a saved agent alias, a URL or a JSON agent config:: + + python -m adcp reporting explain --agent my-seller --account acc_a --period 2026-11-03 + python -m adcp reporting reconcile --agent my-seller [--since 2026-11-01] [--json] + Comparing a buyer's trace with a seller's:: python -m adcp reporting diff buyer.json seller.json @@ -29,7 +34,7 @@ from collections.abc import Sequence from typing import Any -USAGE = "python -m adcp reporting {explain,diff} ..." +USAGE = "python -m adcp reporting {explain,reconcile,diff} ..." def _load_app(target: str) -> Any: @@ -73,9 +78,34 @@ def _print_traces(traces: Sequence[Any], *, as_json: bool) -> None: print(f" {trace.explanation}") +def _agent_client(identifier: str) -> Any: + from adcp.__main__ import resolve_agent_config + from adcp.client import ADCPClient + from adcp.types.core import AgentConfig, Protocol + + config = dict(resolve_agent_config(identifier)) + config.setdefault("id", config.get("agent_uri", "unknown")) + protocol = config.get("protocol", "mcp") + config["protocol"] = Protocol(protocol.lower()) if isinstance(protocol, str) else protocol + return ADCPClient(AgentConfig(**config)) + + async def _explain(args: argparse.Namespace) -> int: from adcp._principal_ref import PrincipalRef + if args.agent: + async with _agent_client(args.agent) as client: + traces = await client.reporting.explain( + account_id=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 + if not args.principal: + raise SystemExit("--app needs --principal") service = await _resolve_app(args.app) traces = await service.explain_coverage( PrincipalRef(args.tenant, args.principal), @@ -89,6 +119,32 @@ async def _explain(args: argparse.Namespace) -> int: return 0 +async def _reconcile(args: argparse.Namespace) -> int: + async with _agent_client(args.agent) as client: + report = await client.reporting.reconcile( + since=args.since, include_not_owed=args.include_not_owed + ) + if args.json: + print(json.dumps(report.to_dict(), indent=2)) + return 0 + summary = report.summary + print( + f"{'definitive' if report.definitive else 'NOT definitive'} as of {report.as_of}: " + f"{summary.owed} owed, {summary.delivered} delivered, {summary.pending} pending, " + f"{summary.late} late, {summary.missing_obligation} missing, " + f"{summary.withheld} withheld" + ) + for issue in report.issues: + print(f" ! {issue.code}: {issue.message}") + for line in report.periods: + print( + f"{line.account_id} {line.subscription_id} v{line.subscription_version} " + f"{line.period['start']}..{line.period['end']} {line.status} [{line.reason}] " + f"{line.responsible_party}" + ) + return 0 + + def _read_traces(path: str) -> list[Any]: from adcp.reporting.subscriptions.explain import CoverageTrace @@ -140,8 +196,10 @@ def main(argv: Sequence[str]) -> int: 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") + source = explain.add_mutually_exclusive_group(required=True) + source.add_argument("--app", help="seller: module:attribute of your service") + source.add_argument("--agent", help="buyer: saved alias, URL or JSON agent config") + explain.add_argument("--principal", default=None, help="seller: 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( @@ -153,12 +211,21 @@ def main(argv: Sequence[str]) -> int: 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") + reconcile = commands.add_parser( + "reconcile", help="List what a seller owes your subscriptions, and each status (buyer)" + ) + reconcile.add_argument("--agent", required=True, help="saved alias, URL or JSON config") + reconcile.add_argument("--since", default=None, help="skip periods starting earlier") + reconcile.add_argument("--include-not-owed", action="store_true") + reconcile.add_argument("--json", action="store_true") 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) + if args.command == "reconcile": + return asyncio.run(_reconcile(args)) return asyncio.run(_explain(args)) diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index 4e8ffb1a7..0dd9751db 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -84,6 +84,7 @@ from adcp.reporting.subscriptions.explain import ( EXPLAIN_TEMPLATES, TRACE_VERSION, + CoverageExplainer, CoverageReason, CoverageTrace, CoverageTraceDifference, @@ -104,6 +105,13 @@ SubscriptionPlanResult, subscription_obligation_id, ) +from adcp.reporting.subscriptions.reconcile import ( + BuyerReportingState, + PeriodLine, + ReconcileIssue, + ReconcileReport, + ReconcileSummary, +) from adcp.reporting.subscriptions.registry import ( ReportingSubscriptionValidationError, apply_reporting_subscriptions, @@ -132,6 +140,12 @@ from adcp.server.reporting_grants import REPORTING_WIND_DOWN_TASKS, wind_down_grant __all__ = [ + "BuyerReportingState", + "CoverageExplainer", + "PeriodLine", + "ReconcileIssue", + "ReconcileReport", + "ReconcileSummary", "EXPLAIN_TEMPLATES", "TRACE_VERSION", "CoverageReason", diff --git a/src/adcp/reporting/subscriptions/consumer.py b/src/adcp/reporting/subscriptions/consumer.py index 770bd68c2..48c04d35b 100644 --- a/src/adcp/reporting/subscriptions/consumer.py +++ b/src/adcp/reporting/subscriptions/consumer.py @@ -317,7 +317,12 @@ class _AccountRecord: class ReportingSubscriptionTracker: - """A buyer's own records for its subscriptions with one seller. + """A buyer's own records for its subscriptions with one seller (advanced). + + Most buyers call ``client.reporting.reconcile()`` and ``explain()`` + instead: they derive the roster and grant history from the wire on each + run, so there is nothing to record. This tracker stays for code that + keeps its own records. Register through :meth:`sync` so every version's configuration is kept. Record what :func:`list_accounts` shows with :meth:`observe_account`. diff --git a/src/adcp/reporting/subscriptions/explain.py b/src/adcp/reporting/subscriptions/explain.py index 90df9760a..0b51d399b 100644 --- a/src/adcp/reporting/subscriptions/explain.py +++ b/src/adcp/reporting/subscriptions/explain.py @@ -1101,201 +1101,242 @@ 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. +def _parse_z(value: str) -> datetime: + return utc_instant(datetime.fromisoformat(value.replace("Z", "+00:00"))) - 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``. +class CoverageExplainer: + """One coverage input, prepared once, explaining as many periods as asked. - Traces are sorted by period start, buy and offering (byte order). A buy - without a match for the selection yields no trace. + :func:`explain_subscription_coverage` is the one-call form; the buyer's + ``reconcile`` uses this to classify every period of an account against + one owed set. The constructor takes the input of + :func:`explain_subscription_coverage` without the selection. """ - 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" + + def __init__( + self, + *, + subscription: Any, + account_id: str, + buys: Sequence[SubscriptionBuy], + offerings: Sequence[SubscriptionOffering], + account_timezone: str | None, + now: datetime, + 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, + ) -> None: + self.now = utc_instant(now, field="now") + if side not in ("buyer", "seller"): + raise ValueError("side is 'buyer' or 'seller'") + self.side: CoverageSide = side + self.hidden = frozenset(invisible_media_buy_ids) + if self.hidden and side != "seller": + raise ValueError("invisible_media_buy_ids is seller-only") + self.principal_id = principal_id + self.seller_only = dict(seller_only or {}) + self.account_id = account_id + self.account_timezone = account_timezone + self.backfill = backfill_on_regrant + self.retention_days = status_retention_days + self.buys = list(buys) + self.offerings = {item.offering_id: item for item in offerings} + context = _coverage_context( + subscription=subscription, + account_id=account_id, + buys=self.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, ) - 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"))), + self.context = context + self.owed: list[OwedSubscriptionPeriod] = [] + if context is not None: + self.owed = owed_subscription_periods( + subscription=context.windows, + account_id=account_id, + buys=[buy for buy in self.buys if buy.media_buy_id not in self.hidden], + offerings=offerings, + account_timezone=account_timezone, + now=self.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, + ) + self._index: dict[tuple[str, str, datetime, datetime, str], OwedSubscriptionPeriod] = {} + self._stubs: dict[tuple[str, str], tuple[datetime, datetime]] = {} + for item in self.owed: + start, end = _parse_z(item.period_start), _parse_z(item.period_end) + for buy_id in item.media_buy_ids: + self._index[(buy_id, item.offering_id, start, end, item.source_timezone)] = item + if item.stub: + self._stubs[(buy_id, item.offering_id)] = (start, end) + self._authorization = _authorization_projection(authorizations, status_retention_days) + + @property + def subscription_id(self) -> str | None: + return None if self.context is None else self.context.subscription_id + + def owed_period( + self, media_buy_id: str, offering_id: str, start: datetime, end: datetime, zone: str + ) -> OwedSubscriptionPeriod | None: + """The owed period (including not yet ended) for one buy, if any.""" + return self._index.get((media_buy_id, offering_id, start, end, zone)) + + def _calendar(self, offering: SubscriptionOffering, zone: str) -> PeriodCalendar: + return 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", ) - 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] - ) + def select( + self, period: Any, *, media_buy_id: str | None = None, offering_id: str | None = None + ) -> list[_Candidate]: + """The (buy, offering, period) candidates a ``period`` selection names.""" + context = self.context + if context is None: + return [] + kind, value = _period_spec(period) + selected = [b for b in self.buys if media_buy_id is None or b.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" ) - 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): + if offering_id is not None and offering_id not in self.offerings: + raise CoverageInputError( + "invalid_input", f"offering_id {offering_id!r} is not among the offerings passed" + ) + zone_default = self.account_timezone or "UTC" + 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 self.offerings] + ) + if not wanted: + if kind == "range": 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}", + elif kind == "instant": + day = value.astimezone(require_zone(zone_default)).date() + periods = [_local_day(day, zone_default)] + else: + periods = [_local_day(value, zone_default)] + candidates.extend( + _Candidate(buy, None, None, s, e, zone_default) for s, e in periods ) + continue + for oid in wanted: + offering = self.offerings[oid] + if kind == "range": + reference = value[0] + elif kind == "instant": + reference = value + else: + reference = _local_day(value, zone_default)[0] + seen: set[tuple[datetime, datetime]] = set() + for zone in _zones(context, offering, [reference, reference + timedelta(days=1)]): + calendar = self._calendar(offering, zone) + periods = _periods_for(calendar, kind, value, zone) + if ( + kind == "range" + and not periods + and self._stubs.get((buy.media_buy_id, oid)) == tuple(value) + ): + periods = [value] + for s, e in periods: + if (s, e) not in seen: + 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}", + ) + return candidates + + def flight_candidates( + self, *, since: datetime | None = None, until: datetime | None = None + ) -> list[_Candidate]: + """Every period of each listed offering that overlaps each buy's flight. + + Bounded by ``[since, until)`` (``until`` defaults to now), so the + buyer's ``reconcile`` can list what is not owed, and why. + """ + context = self.context + if context is None: + return [] + upper_bound = self.now if until is None else utc_instant(until, field="until") + result: list[_Candidate] = [] + for buy in self.buys: + low = buy._earliest_start() + high = min(buy._latest_stop(), upper_bound) + if since is not None: + low = max(low, utc_instant(since, field="since")) + if high <= low: + continue + listed = sorted({oid for _pkg, ids in buy.product_offering_ids() for oid in ids}) + for oid in [o for o in listed if o in self.offerings]: + offering = self.offerings[oid] + for zone in _zones(context, offering, [low, high]): + calendar = self._calendar(offering, zone) + for s, e in calendar.periods( + starts_from=calendar.floor(low), starts_before=high + ): + result.append(_Candidate(buy, offering, calendar, s, e, zone)) + return result - authorization = _authorization_projection(authorizations, status_retention_days) - - traces: list[CoverageTrace] = [] - for candidate in candidates: + def trace(self, candidate: _Candidate) -> CoverageTrace: + """The trace of one candidate.""" + context = self.context + assert context is not None 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, - ) + buy_id = candidate.buy.media_buy_id + owed_item = self._index.get( + (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 + for (owed_buy, oid, s, e, _zone), item in self._index.items() + if owed_buy == buy_id and oid == key_offering and s < candidate.end and e > candidate.start @@ -1307,17 +1348,17 @@ def explain_subscription_coverage( candidate, owed_item, elsewhere, - now=now, - backfill=backfill_on_regrant, - invisible=candidate.buy.media_buy_id in hidden, + now=self.now, + backfill=self.backfill, + invisible=buy_id in self.hidden, ) assert walk.reason is not None reason = _reason(walk.reason) subject: dict[str, Any] = { - "account_id": account_id, + "account_id": self.account_id, "subscription_id": context.subscription_id, "subscription_version": version, - "media_buy_id": candidate.buy.media_buy_id, + "media_buy_id": buy_id, "offering_id": candidate.offering.offering_id if candidate.offering else None, "period": { "start": iso_z(candidate.start), @@ -1325,43 +1366,89 @@ def explain_subscription_coverage( "timezone": candidate.zone, }, } - if principal_id is not None: - subject = {"principal_id": principal_id, **subject} + if self.principal_id is not None: + subject = {"principal_id": self.principal_id, **subject} inputs = _inputs_document( context, candidate, subject, - authorization=authorization, - backfill=backfill_on_regrant, - status_retention_days=status_retention_days, + authorization=self._authorization, + backfill=self.backfill, + status_retention_days=self.retention_days, ) only = None - if side == "seller": - only = {"buy_visible": candidate.buy.media_buy_id not in hidden, **(seller_only or {})} - traces.append( - CoverageTrace( - side=side, - subject=subject, - verdict=reason.verdict, - reason=reason.code, - explanation=_explanation(reason.code, subject, walk.params), - steps=tuple(walk.steps), - inputs_digest=_digest(inputs), - seller_only=only, - ) + if self.side == "seller": + only = {"buy_visible": buy_id not in self.hidden, **self.seller_only} + return CoverageTrace( + side=self.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( + + def explain( + self, period: Any, *, media_buy_id: str | None = None, offering_id: str | None = None + ) -> list[CoverageTrace]: + traces = [ + self.trace(candidate) + for candidate in self.select(period, media_buy_id=media_buy_id, offering_id=offering_id) + ] + return sort_traces(traces) + + +def sort_traces(traces: Iterable[CoverageTrace]) -> list[CoverageTrace]: + """Traces by period start, buy and offering (byte order), then zone.""" + return sorted( + traces, key=lambda t: ( t.subject["period"]["start"], - str(t.subject["media_buy_id"]).encode(), - str(t.subject["offering_id"] or "").encode(), + str(t.subject.get("media_buy_id")).encode(), + str(t.subject.get("offering_id") or "").encode(), t.subject["period"]["timezone"], - ) + ), ) - return traces + + +def explain_subscription_coverage( + *, + period: Any, + media_buy_id: str | None = None, + offering_id: str | None = None, + **inputs: Any, +) -> list[CoverageTrace]: + """One :class:`CoverageTrace` per (buy, offering, period) the request selects. + + Takes the input of :func:`owed_subscription_periods` (``subscription``, + ``account_id``, ``buys``, ``offerings``, ``account_timezone``, ``now`` + and the optional facts), 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 passed when omitted. + :param offering_id: one offering; every offering the buy's packages list + when omitted. A buy that lists no offering yields one trace without + ``offering_id``. + :param side: ``buyer`` (default) 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``. + :param principal_id: the seller's name for the caller, echoed in the subject. + + Traces are sorted by period start, buy and offering (byte order). A buy + without a match for the selection yields no trace. + """ + explainer = CoverageExplainer(**inputs) + return explainer.explain(period, media_buy_id=media_buy_id, offering_id=offering_id) __all__ = [ + "CoverageExplainer", + "sort_traces", "EXPLAIN_TEMPLATES", "EXPLAIN_TEMPLATES_VERSION", "TRACE_VERSION", diff --git a/src/adcp/reporting/subscriptions/facade.py b/src/adcp/reporting/subscriptions/facade.py index 5c504c8a1..76714f24c 100644 --- a/src/adcp/reporting/subscriptions/facade.py +++ b/src/adcp/reporting/subscriptions/facade.py @@ -21,7 +21,8 @@ from collections.abc import Awaitable, Callable, Mapping, Sequence from dataclasses import dataclass, field -from typing import Any, Literal +from datetime import datetime, timedelta +from typing import TYPE_CHECKING, Any, Literal from uuid import uuid4 from adcp.reporting.subscriptions.consumer import ( @@ -33,6 +34,13 @@ from adcp.reporting.subscriptions.registry import _semantic from adcp.types import GetPrincipalRequest, SyncPrincipalRequest +if TYPE_CHECKING: + from adcp.reporting.subscriptions.explain import CoverageTrace + from adcp.reporting.subscriptions.reconcile import ReconcileReport + +#: The default ``grace`` of :meth:`ReportingClient.reconcile`: one hour, never zero. +DEFAULT_GRACE = timedelta(hours=1) + SubscribeAction = Literal["created", "updated", "unchanged", "reactivated", "deactivated"] #: Conflict retries with a fresh key, and the bound on the roster read. @@ -265,6 +273,9 @@ def __init__( self._client = client self._attempts = max_attempts self._key = idempotency_key or (lambda: f"rsub-{uuid4()}") + from adcp.reporting.subscriptions.reconcile import BuyerReportingState + + self._state = BuyerReportingState() # -- public API --------------------------------------------------------------- @@ -384,6 +395,84 @@ async def build(principal: _Principal) -> _Plan: plan_result = await self._write(build, single=False) return plan_result.readback + async def reconcile( + self, + *, + since: Any = None, + accounts: Sequence[str] | Literal["covered"] = "covered", + subscription_id: str | None = None, + grace: timedelta = DEFAULT_GRACE, + include_not_owed: bool = False, + now: datetime | None = None, + ) -> ReconcileReport: + """What the seller owes every covered account, and whose move each period is. + + Reads everything from the wire on each run: the roster is + ``list_accounts`` ∩ the subscriptions' scopes plus their + ``accounts_unauthorized[]``; grant history is ``authorized_at``, + ``regranted_at`` and ``unauthorized_at``. Each owed period that has + ended becomes a :class:`~adcp.reporting.subscriptions.reconcile.PeriodLine`: + ``delivered``, ``pending``, ``late`` (the revision is ``grace`` past + ``expected_at``), ``missing_obligation`` (no obligation ``grace`` after + the period ended), or ``withheld`` (authority lost). ``reason`` is the + spec reason code and ``responsible_party`` says who acts. + + :param since: a date or instant; earlier periods are skipped. The + default is all retained coverage. + :param accounts: ``"covered"`` (the derived roster) or explicit ids. + :param grace: how long the seller may take after a period end, or after + ``expected_at``, before a missing obligation or revision counts. + Defaults to one hour, never zero. + :param include_not_owed: add a ``not_owed`` line, with its reason, for + every other period of each buy's flight (the audit trail). + + ``report.definitive`` is true only when every read completed; + otherwise ``report.issues`` say what to retry. + """ + from adcp.reporting.subscriptions.reconcile import reconcile_subscriptions + + return await reconcile_subscriptions( + self._client, + state=self._state, + since=since, + accounts=accounts, + subscription_id=subscription_id, + grace=grace, + include_not_owed=include_not_owed, + now=now, + ) + + async def explain( + self, + *, + account_id: str, + period: Any, + subscription_id: str | None = None, + media_buy_id: str | None = None, + offering_id: str | None = None, + now: datetime | None = None, + ) -> list[CoverageTrace]: + """Why each selected period is, or is not, owed: the buyer's ``CoverageTrace`` list. + + The same trace the seller's ``explain_coverage`` builds, with + ``side="buyer"``, from the facts this client reads. ``period`` is a + ``YYYY-MM-DD`` local date read in each offering's zone, an instant, or + ``{"start", "end"}``. Compare with a seller trace through + :func:`~adcp.reporting.subscriptions.diff_coverage_traces`. + """ + from adcp.reporting.subscriptions.reconcile import explain_buyer_coverage + + return await explain_buyer_coverage( + self._client, + account_id=account_id, + period=period, + state=self._state, + subscription_id=subscription_id, + media_buy_id=media_buy_id, + offering_id=offering_id, + now=now, + ) + # -- planning ------------------------------------------------------------------- async def _plan_subscribe( diff --git a/src/adcp/reporting/subscriptions/reconcile.py b/src/adcp/reporting/subscriptions/reconcile.py new file mode 100644 index 000000000..e264f32ff --- /dev/null +++ b/src/adcp/reporting/subscriptions/reconcile.py @@ -0,0 +1,899 @@ +"""The buyer's ``reconcile`` and ``explain``: what am I owed, and why (AdCP 3.3, experimental). + +``client.reporting.reconcile()`` answers, for every account the caller's +subscriptions cover, which periods the seller owes, which it delivered, and +whose move each one is. Everything is read from the wire on each run: + +* the roster is ``list_accounts`` ∩ the subscriptions' scopes, plus each + subscription's ``accounts_unauthorized[]``, so there is no static account + list to keep; +* the subscriptions, their versions (``prior_versions[].configuration``) and + ``inactive_intervals`` come from ``get_principal``; +* each account's ``authorization.authorized_at`` and ``regranted_at``, its + ``timezone`` and ``operator`` come from ``list_accounts``; a loss comes from + ``accounts_unauthorized[].unauthorized_at``. Grant history is never the + caller's to record; +* buys come from ``get_media_buys``, offerings, ``backfill_on_regrant`` and + ``status_retention_days`` from capabilities, ``subscription_coverage[]`` and + the obligations from a complete ``get_reporting_status`` periods walk. + +The owed set is :func:`~adcp.reporting.subscriptions.owed_subscription_periods` +over those facts, the same rule the seller runs; each period becomes a +:class:`PeriodLine` with a ``status``, the spec ``reason`` code and the +``responsible_party``. ``explain`` returns the same +:class:`~adcp.reporting.subscriptions.CoverageTrace` the seller's +``explain_coverage`` builds, with ``side="buyer"``. + +The client keeps one cache, per account: the last facts read while the caller +was authorized (``authorized_at``, timezone, operator, buys and coverage). It +is what lets ``reconcile`` show an account whose authority was lost, which the +seller then reports as unknown, with its owed periods ``withheld``. +""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass, field +from datetime import date, datetime, timedelta, timezone +from typing import Any, Literal + +from adcp.reporting.subscriptions._clock import CoverageInputError, iso_z, utc_instant +from adcp.reporting.subscriptions.coverage import ( + OwedSubscriptionPeriod, + SubscriptionAuthorization, + SubscriptionBuy, + subscription_covers_account, + subscription_version_windows, +) +from adcp.reporting.subscriptions.explain import ( + CoverageExplainer, + CoverageTrace, + _Candidate, + sort_traces, +) +from adcp.reporting.subscriptions.resolution import SubscriptionOffering + +PeriodStatus = Literal["delivered", "pending", "late", "missing_obligation", "withheld", "not_owed"] +ResponsibleParty = Literal["seller", "buyer", "none"] + +#: How long after a period's end, or its ``expected_at``, before a missing +#: obligation or revision counts against the seller. +DEFAULT_GRACE = timedelta(hours=1) +#: Pages walked for one roster or buy listing; a walk that does not end is +#: refused, never truncated. +_MAX_PAGES = 1_000 +_BUY_STATUSES = ( + "pending_creatives", + "pending_start", + "active", + "paused", + "completed", + "rejected", + "canceled", +) +_FINALITY_RANK = {"provisional": 0, "snapshot": 1, "official": 2} + + +@dataclass(frozen=True) +class PeriodLine: + """One period of one (subscription version, offering) on one account. + + Keys are wire-style in both SDKs, so exports match. ``reason`` is a spec + reason code and ``message`` its explanation sentence. + """ + + account_id: str + subscription_id: str + subscription_version: int + offering_id: str | None + report_definition_id: str | None + media_buy_ids: tuple[str, ...] + period: Mapping[str, str] + status: PeriodStatus + reason: str + responsible_party: ResponsibleParty + message: str + expected_at: str | None = None + reporting_obligation_id: str | None = None + reporting_revision_id: str | None = None + finality: str | None = None + receipt: Mapping[str, Any] | None = None + + def to_dict(self) -> dict[str, Any]: + result: dict[str, Any] = { + "account_id": self.account_id, + "subscription_id": self.subscription_id, + "subscription_version": self.subscription_version, + "offering_id": self.offering_id, + "report_definition_id": self.report_definition_id, + "media_buy_ids": list(self.media_buy_ids), + "period": dict(self.period), + "status": self.status, + "reason": self.reason, + "responsible_party": self.responsible_party, + "message": self.message, + } + for name in ( + "expected_at", + "reporting_obligation_id", + "reporting_revision_id", + "finality", + ): + value = getattr(self, name) + if value is not None: + result[name] = value + if self.receipt is not None: + result["receipt"] = dict(self.receipt) + return result + + +@dataclass(frozen=True) +class ReconcileSummary: + """Counts by status; ``owed`` is every line except ``not_owed``.""" + + owed: int = 0 + delivered: int = 0 + pending: int = 0 + late: int = 0 + missing_obligation: int = 0 + withheld: int = 0 + not_owed: int = 0 + + def to_dict(self) -> dict[str, int]: + return dict(vars(self)) + + +@dataclass(frozen=True) +class ReconcileIssue: + """Why part of a reconcile is not definitive; the message ends with the fix.""" + + code: str + message: str + account_id: str | None = None + subscription_id: str | None = None + + +@dataclass(frozen=True) +class ReconcileReport: + """The result of ``client.reporting.reconcile()``. + + ``definitive`` is true only when the roster and every account's + obligations walk completed and every owed set was computed; otherwise + ``issues`` say what is missing, and the lines are what could be read, + never a smaller success. + """ + + as_of: str + definitive: bool + summary: ReconcileSummary + periods: tuple[PeriodLine, ...] + issues: tuple[ReconcileIssue, ...] = () + accounts: tuple[str, ...] = () + + def to_dict(self) -> dict[str, Any]: + return { + "as_of": self.as_of, + "definitive": self.definitive, + "summary": self.summary.to_dict(), + "periods": [line.to_dict() for line in self.periods], + "issues": [vars(issue) for issue in self.issues], + "accounts": list(self.accounts), + } + + +# -- wire reads ---------------------------------------------------------------------- + + +def _plain(value: Any) -> Any: + if hasattr(value, "model_dump"): + return value.model_dump(mode="json", exclude_none=True, by_alias=True) + if isinstance(value, Mapping): + return {key: _plain(item) for key, item in value.items() if item is not None} + if isinstance(value, (list, tuple)): + return [_plain(item) for item in value] + return value + + +def _data(result: Any, task: str) -> Any: + data = getattr(result, "data", None) + if not getattr(result, "success", False) or data is None: + raise _UnreadableError(task, str(getattr(result, "error", None) or "no response")) + return data + + +class _UnreadableError(Exception): + def __init__(self, task: str, detail: str) -> None: + super().__init__(f"{task} failed: {detail}") + self.task = task + + +def _instant(value: Any) -> datetime | None: + if value is None: + return None + if isinstance(value, datetime): + return utc_instant(value) + from adcp.reporting._timestamp import aware_timestamp + + return utc_instant(aware_timestamp(str(value))) + + +@dataclass +class _Account: + """What the caller last read for one account while authorized.""" + + account_id: str + timezone: str | None = None + operator: str | None = None + authorized_at: datetime | None = None + regranted_at: datetime | None = None + #: ``unauthorized_at`` values seen in ``accounts_unauthorized[]``. + losses: list[datetime] = field(default_factory=list) + buys: list[dict[str, Any]] = field(default_factory=list) + #: subscription_id -> the ``subscription_coverage[]`` entry + coverage: dict[str, dict[str, Any]] = field(default_factory=dict) + listed: bool = False + + +@dataclass +class _Seller: + offerings: tuple[SubscriptionOffering, ...] + backfill_on_regrant: bool + status_retention_days: int | None + + +class BuyerReportingState: + """The buyer's cache of the last facts read per account (in memory). + + Only what the wire stops showing after a loss of authority is kept: + the chain start, zone, operator, buys and coverage last read. + """ + + def __init__(self) -> None: + self.accounts: dict[str, _Account] = {} + + def account(self, account_id: str) -> _Account: + record = self.accounts.get(account_id) + if record is None: + record = self.accounts[account_id] = _Account(account_id) + return record + + +async def _seller_facts(client: Any) -> _Seller: + fetch = getattr(client, "fetch_capabilities", None) + if fetch is None: + raise _UnreadableError("get_adcp_capabilities", "the client cannot read capabilities") + capabilities = await fetch() + payload = _plain(capabilities) + block = (payload.get("media_buy") or {}).get("reporting_delivery") or {} + offerings = [] + for item in block.get("offerings") or (): + try: + offerings.append(SubscriptionOffering.from_offering(item)) + except (TypeError, ValueError): + continue # an offering without standard_dimensions cannot be selected + subscriptions = block.get("subscriptions") or {} + return _Seller( + offerings=tuple(offerings), + backfill_on_regrant=bool(subscriptions.get("backfill_on_regrant", False)), + status_retention_days=block.get("status_retention_days"), + ) + + +async def _roster(client: Any) -> list[dict[str, Any]]: + from adcp.types import ListAccountsRequest + + list_accounts = getattr(client, "list_accounts", None) + if list_accounts is None: + raise _UnreadableError("list_accounts", "the client cannot list accounts") + accounts: list[dict[str, Any]] = [] + cursor: str | None = None + for _page in range(_MAX_PAGES): + payload: dict[str, Any] = {} + if cursor: + payload["pagination"] = {"cursor": cursor} + data = _plain( + _data(await list_accounts(ListAccountsRequest.model_validate(payload)), "list_accounts") + ) + accounts.extend(data.get("accounts") or ()) + pagination = data.get("pagination") or {} + cursor = pagination.get("cursor") if pagination.get("has_more") else None + if not cursor: + return accounts + raise _UnreadableError("list_accounts", f"did not finish paging within {_MAX_PAGES} pages") + + +async def _buys(client: Any, account_id: str) -> list[dict[str, Any]]: + from adcp.types import GetMediaBuysRequest + + buys: list[dict[str, Any]] = [] + cursor: str | None = None + for _page in range(_MAX_PAGES): + payload: dict[str, Any] = { + "account": {"account_id": account_id}, + "status_filter": list(_BUY_STATUSES), + } + if cursor: + payload["pagination"] = {"cursor": cursor} + data = _plain( + _data( + await client.get_media_buys(GetMediaBuysRequest.model_validate(payload)), + "get_media_buys", + ) + ) + buys.extend(data.get("media_buys") or ()) + pagination = data.get("pagination") or {} + cursor = pagination.get("cursor") if pagination.get("has_more") else None + if not cursor: + return buys + raise _UnreadableError("get_media_buys", f"did not finish paging within {_MAX_PAGES} pages") + + +async def _coverage(client: Any, account_id: str) -> dict[str, dict[str, Any]]: + from adcp.types import GetReportingStatusRequest + + request = GetReportingStatusRequest.model_validate( + {"account": {"account_id": account_id}, "view": "summary"} + ) + data = _plain(_data(await client.get_reporting_status(request), "get_reporting_status")) + return { + str(entry["subscription_id"]): entry for entry in data.get("subscription_coverage") or () + } + + +def _subscription_buys( + raw: Iterable[Mapping[str, Any]], every: Sequence[str] +) -> list[SubscriptionBuy]: + """Buys as coverage reads them. + + Each package's ``reporting_delivery_offering_ids`` snapshot is used as + served (``[]`` means no offering). A package served without one is read + as every advertised offering, the same fallback the SDK seller applies, + so both sides compute the same owed set. + """ + result = [] + for item in raw: + if item.get("start_time") is None or item.get("end_time") is None: + continue # not yet flighted: nothing can be owed for it + media_buy = dict(item) + media_buy["packages"] = [ + { + **package, + "reporting_delivery_offering_ids": list( + package["reporting_delivery_offering_ids"] + if isinstance(package.get("reporting_delivery_offering_ids"), (list, tuple)) + else every + ), + } + for package in item.get("packages") or () + if package.get("package_id") + ] + buy = SubscriptionBuy.from_media_buy(media_buy) + if not buy.packages: + from dataclasses import replace + + buy = replace(buy, reporting_delivery_offering_ids=tuple(every)) + result.append(buy) + return result + + +def _grants( + record: _Account, coverage: Mapping[str, Any] | None +) -> list[SubscriptionAuthorization]: + """The grant history the wire states for one account. + + ``authorized_at`` starts the current chain and ``regranted_at`` is its + latest re-grant; the loss before that re-grant is the + ``unauthorized_at`` last seen in ``accounts_unauthorized[]``, else the + start of the seller's ``gaps[]`` entry for it (the last boundary at or + before the loss, which removes exactly the same periods). A current loss + ends the chain. + """ + start = record.authorized_at + if start is None: + return [] + regranted = record.regranted_at + current_loss = None if record.listed else max(record.losses, default=None) + if regranted is None or regranted <= start: + if current_loss is not None and current_loss > start: + return [SubscriptionAuthorization(start, current_loss)] + return [SubscriptionAuthorization(start)] + seen = [loss for loss in record.losses if start < loss < regranted] + if seen: + lost = max(seen) + else: + gap_starts = [ + _instant(gap.get("from")) + for gap in (coverage or {}).get("gaps") or () + if (_instant(gap.get("until")) or regranted) >= regranted + ] + candidates = [item for item in gap_starts if item is not None and start < item < regranted] + lost = max(candidates) if candidates else regranted - timedelta(microseconds=1) + lost = max(lost, start + timedelta(microseconds=1)) + later = SubscriptionAuthorization( + regranted, current_loss if current_loss and current_loss > regranted else None + ) + return [SubscriptionAuthorization(start, lost), later] + + +def _configuration(state: Any) -> dict[str, Any]: + return dict(_plain(getattr(state, "configuration", None) or {})) + + +def _scope_covers(state: Any, record: _Account) -> bool: + configuration = _configuration(state) + return subscription_covers_account( + configuration, account_id=record.account_id, account_operator=record.operator + ) + + +class _Context: + """Every wire fact one reconcile or explain run reads.""" + + def __init__(self, client: Any, state: BuyerReportingState) -> None: + self.client = client + self.state = state + self.issues: list[ReconcileIssue] = [] + self.subscriptions: tuple[Any, ...] = () + self.seller: _Seller | None = None + self.roster_complete = False + + def issue( + self, + code: str, + message: str, + *, + account_id: str | None = None, + subscription_id: str | None = None, + ) -> None: + self.issues.append(ReconcileIssue(code, message, account_id, subscription_id)) + + async def load(self, *, accounts: Sequence[str] | None) -> list[_Account]: + from adcp.reporting.subscriptions.facade import read_principal + + principal = await read_principal(self.client) + self.subscriptions = principal.subscriptions + try: + self.seller = await _seller_facts(self.client) + except Exception as error: # noqa: BLE001 - reported, never a smaller success + self.issue( + "SUBSCRIPTION_OFFERINGS_UNAVAILABLE", + f"Capabilities could not be read ({error}). Retry once the seller is reachable.", + ) + self.seller = _Seller((), False, None) + unauthorized: dict[str, datetime] = {} + for state in self.subscriptions: + for entry in getattr(state, "accounts_unauthorized", None) or (): + at = _instant(getattr(entry, "unauthorized_at", None)) + if at is not None: + previous = unauthorized.get(entry.account_id) + unauthorized[entry.account_id] = max(at, previous) if previous else at + listed = getattr(state, "accounts_unauthorized_count", None) + inline = len(getattr(state, "accounts_unauthorized", None) or ()) + if listed is not None and listed > inline: + self.issue( + "SUBSCRIPTION_ROSTER_UNAVAILABLE", + f"{state.subscription_id} lists {inline} of {listed} unauthorized accounts " + "inline. Page the rest through get_principal reporting_subscription_page.", + subscription_id=state.subscription_id, + ) + try: + roster = await _roster(self.client) + self.roster_complete = True + except Exception as error: # noqa: BLE001 + roster = [] + self.issue( + "SUBSCRIPTION_ROSTER_UNAVAILABLE", + f"The account roster could not be read completely ({error}); a partial roster " + "is never used. Retry once list_accounts answers.", + ) + seen: set[str] = set() + for raw in roster: + account_id = str(raw.get("account_id") or "") + if not account_id: + continue + record = self.state.account(account_id) + record.listed = True + seen.add(account_id) + record.timezone = raw.get("timezone") or record.timezone + record.operator = raw.get("operator") or record.operator + authorization = raw.get("authorization") or {} + record.authorized_at = _instant(authorization.get("authorized_at")) + record.regranted_at = _instant(authorization.get("regranted_at")) + for account_id, lost in unauthorized.items(): + record = self.state.account(account_id) + if account_id not in seen: + record.listed = False + if lost not in record.losses: + record.losses.append(lost) + if self.roster_complete: + for record in self.state.accounts.values(): + if record.account_id not in seen and record.account_id not in unauthorized: + record.listed = False + wanted = set(accounts) if accounts is not None else None + result: list[_Account] = [] + for record in sorted(self.state.accounts.values(), key=lambda r: r.account_id): + if wanted is not None and record.account_id not in wanted: + continue + covered = any(_scope_covers(state, record) for state in self.subscriptions) + if not covered and record.account_id not in unauthorized: + continue + if not record.listed and record.account_id not in unauthorized: + continue + result.append(record) + if wanted is not None: + for missing in sorted(wanted - {r.account_id for r in result}): + self.issue( + "SUBSCRIPTION_ACCOUNT_NOT_COVERED", + f"Account {missing!r} is not readable or no subscription covers it. " + "Check list_accounts and the subscription scopes.", + account_id=missing, + ) + return result + + async def refresh(self, record: _Account) -> bool: + """Read buys and coverage for a readable account; ``False`` when it is not.""" + if not record.listed: + return False + try: + record.buys = await _buys(self.client, record.account_id) + record.coverage = await _coverage(self.client, record.account_id) + except Exception as error: # noqa: BLE001 + self.issue( + "SUBSCRIPTION_ACCOUNT_UNREADABLE", + f"Account {record.account_id!r} could not be read ({error}). The last facts " + "read for it are used; retry once the seller answers.", + account_id=record.account_id, + ) + return False + return True + + def explainer(self, state: Any, record: _Account, now: datetime) -> CoverageExplainer: + assert self.seller is not None + coverage = record.coverage.get(str(state.subscription_id)) + windows = subscription_version_windows(state, require_prior_configurations=False) + return CoverageExplainer( + subscription=windows, + account_id=record.account_id, + buys=_subscription_buys( + record.buys, [item.offering_id for item in self.seller.offerings] + ), + offerings=self.seller.offerings, + account_timezone=record.timezone, + now=now, + authorizations=_grants(record, coverage), + coverage=coverage, + backfill_on_regrant=self.seller.backfill_on_regrant, + status_retention_days=self.seller.status_retention_days, + account_operator=record.operator, + ) + + +# -- reconcile ------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class _Obligation: + obligation_id: str + revision_id: str | None + finality: str | None + receipt_owed: bool + + +async def _obligations( + client: Any, account_id: str, now: datetime +) -> tuple[dict[tuple[Any, ...], _Obligation], bool]: + """Obligations keyed by (subscription, version, offering, start, end), and walk completeness.""" + from adcp.reporting._reconcile import evaluate_reporting_ledger, load_reporting_ledger + from adcp.types import GetReportingStatusRequest + + request = GetReportingStatusRequest.model_validate( + {"account": {"account_id": account_id}, "view": "periods"} + ) + ledger = await load_reporting_ledger(client, request) + result = evaluate_reporting_ledger(ledger, now=now) + revisions = {item.reporting_revision_id: item for item in ledger.revisions} + outcomes = {item.reporting_obligation_id: item for item in result.obligations} + index: dict[tuple[Any, ...], _Obligation] = {} + for obligation in ledger.obligations: + outcome = outcomes.get(obligation.reporting_obligation_id) + revision = ( + revisions.get(outcome.reporting_revision_id) + if outcome and outcome.reporting_revision_id + else None + ) + finality = ( + None + if revision is None + else str(getattr(revision.finality, "value", revision.finality)) + ) + key = ( + obligation.delivery_config_id, + obligation.delivery_config_version, + obligation.offering_id, + utc_instant(obligation.period.start), + utc_instant(obligation.period.end), + ) + index[key] = _Obligation( + obligation.reporting_obligation_id, + None if revision is None else revision.reporting_revision_id, + finality, + bool(outcome and "MISSING_MATCHING_CONSUMER_RECEIPT" in outcome.reasons), + ) + return index, ledger._read_fingerprint is not None + + +def _line( + period: OwedSubscriptionPeriod, + trace_reason: str, + message: str, + obligation: _Obligation | None, + required_finality: str, + *, + now: datetime, + grace: timedelta, +) -> PeriodLine: + end = utc_instant(datetime.fromisoformat(period.period_end.replace("Z", "+00:00"))) + expected = ( + None + if period.expected_at is None + else utc_instant(datetime.fromisoformat(period.expected_at.replace("Z", "+00:00"))) + ) + status: PeriodStatus + responsible: ResponsibleParty = "none" + receipt = None + if ( + obligation is not None + and obligation.revision_id is not None + and ( + _FINALITY_RANK.get(obligation.finality or "", -1) + >= _FINALITY_RANK.get(required_finality, 2) + ) + ): + status = "delivered" + if obligation.receipt_owed: + responsible = "buyer" + receipt = {"status": "owed_by_buyer"} + elif period.withheld: + status = "withheld" + elif obligation is None: + if now >= end + grace: + status, responsible = "missing_obligation", "seller" + else: + status = "pending" + elif expected is not None and now >= expected + grace: + status, responsible = "late", "seller" + else: + status = "pending" + return PeriodLine( + account_id=period.account_id, + subscription_id=period.subscription_id, + subscription_version=period.delivery_config_version, + offering_id=period.offering_id, + report_definition_id=period.report_definition_id, + media_buy_ids=tuple(period.media_buy_ids), + period={ + "start": period.period_start, + "end": period.period_end, + "timezone": period.source_timezone, + }, + status=status, + reason=trace_reason, + responsible_party=responsible, + message=message, + expected_at=period.expected_at, + reporting_obligation_id=None if obligation is None else obligation.obligation_id, + reporting_revision_id=None if obligation is None else obligation.revision_id, + finality=None if obligation is None else obligation.finality, + receipt=receipt, + ) + + +def _owed_reason(explainer: CoverageExplainer, period: OwedSubscriptionPeriod) -> tuple[str, str]: + """The trace reason and sentence of an owed, ended period.""" + start = utc_instant(datetime.fromisoformat(period.period_start.replace("Z", "+00:00"))) + end = utc_instant(datetime.fromisoformat(period.period_end.replace("Z", "+00:00"))) + buy_id = period.media_buy_ids[0] + buy = next(b for b in explainer.buys if b.media_buy_id == buy_id) + offering = explainer.offerings[period.offering_id] + candidate = _Candidate( + buy, + offering, + explainer._calendar(offering, period.source_timezone), + start, + end, + period.source_timezone, + ) + trace = explainer.trace(candidate) + return trace.reason, trace.explanation + + +def _since(value: Any) -> datetime | None: + if value is None: + return None + if isinstance(value, datetime): + return utc_instant(value, field="since") + if isinstance(value, date): + return datetime(value.year, value.month, value.day, tzinfo=timezone.utc) + text = str(value) + if len(text) == 10: + day = date.fromisoformat(text) + return datetime(day.year, day.month, day.day, tzinfo=timezone.utc) + instant = _instant(text) + assert instant is not None + return instant + + +async def reconcile_subscriptions( + client: Any, + *, + state: BuyerReportingState | None = None, + since: Any = None, + accounts: Sequence[str] | Literal["covered"] = "covered", + subscription_id: str | None = None, + grace: timedelta = DEFAULT_GRACE, + include_not_owed: bool = False, + now: datetime | None = None, +) -> ReconcileReport: + """What every covered account is owed, and its status. See :meth:`ReportingClient.reconcile`.""" + instant = utc_instant(now, field="now") if now is not None else datetime.now(timezone.utc) + if not isinstance(grace, timedelta) or grace < timedelta(0): + raise ValueError("grace must be a non-negative timedelta") + lower = _since(since) + context = _Context(client, state or BuyerReportingState()) + records = await context.load(accounts=None if accounts == "covered" else list(accounts)) + lines: list[PeriodLine] = [] + complete = context.roster_complete + for record in records: + readable = await context.refresh(record) + obligations: dict[tuple[Any, ...], _Obligation] = {} + if readable: + try: + obligations, walked = await _obligations(client, record.account_id, instant) + complete = complete and walked + except Exception as error: # noqa: BLE001 + complete = False + context.issue( + "SUBSCRIPTION_LEDGER_UNAVAILABLE", + f"The obligations of {record.account_id!r} could not be walked completely " + f"({error}); periods are reported without delivery status. Retry later.", + account_id=record.account_id, + ) + for state_item in context.subscriptions: + sub_id = str(state_item.subscription_id) + if subscription_id is not None and sub_id != subscription_id: + continue + if not _scope_covers(state_item, record) and not any( + entry.account_id == record.account_id + for entry in getattr(state_item, "accounts_unauthorized", None) or () + ): + continue + try: + explainer = context.explainer(state_item, record, instant) + except CoverageInputError as error: + complete = False + context.issue( + error.code, + f"{sub_id} on {record.account_id!r}: {error}", + account_id=record.account_id, + subscription_id=sub_id, + ) + continue + selector = _configuration(state_item).get("selector") or {} + required = str(selector.get("required_finality") or "official") + for period in explainer.owed: + start = utc_instant( + datetime.fromisoformat(period.period_start.replace("Z", "+00:00")) + ) + end = utc_instant(datetime.fromisoformat(period.period_end.replace("Z", "+00:00"))) + if end > instant or (lower is not None and start < lower): + continue + reason, message = _owed_reason(explainer, period) + obligation = obligations.get( + (sub_id, period.delivery_config_version, period.offering_id, start, end) + ) + lines.append( + _line(period, reason, message, obligation, required, now=instant, grace=grace) + ) + if include_not_owed: + for candidate in explainer.flight_candidates(since=lower, until=instant): + if candidate.end > instant: + continue + trace = explainer.trace(candidate) + if trace.verdict == "owed": + continue + lines.append(_not_owed_line(trace)) + lines.sort( + key=lambda line: ( + line.account_id.encode(), + line.subscription_id.encode(), + line.period["start"], + str(line.offering_id or "").encode(), + ",".join(line.media_buy_ids).encode(), + ) + ) + counts = {name: 0 for name in ReconcileSummary.__dataclass_fields__} + for line in lines: + counts[line.status] += 1 + if line.status != "not_owed": + counts["owed"] += 1 + return ReconcileReport( + as_of=iso_z(instant), + definitive=complete and not context.issues, + summary=ReconcileSummary(**counts), + periods=tuple(lines), + issues=tuple(context.issues), + accounts=tuple(record.account_id for record in records), + ) + + +def _not_owed_line(trace: CoverageTrace) -> PeriodLine: + subject = trace.subject + return PeriodLine( + account_id=subject["account_id"], + subscription_id=subject["subscription_id"], + subscription_version=int(subject["subscription_version"]), + offering_id=subject.get("offering_id"), + report_definition_id=None, + media_buy_ids=(subject["media_buy_id"],), + period=dict(subject["period"]), + status="not_owed", + reason=trace.reason, + responsible_party="none", + message=trace.explanation, + ) + + +async def explain_buyer_coverage( + client: Any, + *, + account_id: str, + period: Any, + state: BuyerReportingState | None = None, + subscription_id: str | None = None, + media_buy_id: str | None = None, + offering_id: str | None = None, + now: datetime | None = None, +) -> list[CoverageTrace]: + """The buyer's :class:`CoverageTrace` list. See :meth:`ReportingClient.explain`.""" + instant = utc_instant(now, field="now") if now is not None else datetime.now(timezone.utc) + context = _Context(client, state or BuyerReportingState()) + records = await context.load(accounts=[account_id]) + record = next((item for item in records if item.account_id == account_id), None) + if record is None: + record = context.state.accounts.get(account_id) + if record is None: + raise CoverageInputError( + "invalid_input", + f"Account {account_id!r} is not readable and no subscription covers it. " + "Check list_accounts and the subscription scopes.", + ) + await context.refresh(record) + traces: list[CoverageTrace] = [] + found = False + for state_item in context.subscriptions: + if subscription_id is not None and str(state_item.subscription_id) != subscription_id: + continue + found = True + explainer = context.explainer(state_item, record, instant) + if media_buy_id is not None and media_buy_id not in { + b.media_buy_id for b in explainer.buys + }: + continue + traces.extend(explainer.explain(period, media_buy_id=media_buy_id, offering_id=offering_id)) + if subscription_id is not None and not found: + raise KeyError(subscription_id) + return sort_traces(traces) + + +__all__ = [ + "DEFAULT_GRACE", + "BuyerReportingState", + "PeriodLine", + "PeriodStatus", + "ReconcileIssue", + "ReconcileReport", + "ReconcileSummary", + "ResponsibleParty", + "explain_buyer_coverage", + "reconcile_subscriptions", +] diff --git a/tests/reporting_inprocess.py b/tests/reporting_inprocess.py new file mode 100644 index 000000000..0f214e68c --- /dev/null +++ b/tests/reporting_inprocess.py @@ -0,0 +1,58 @@ +"""A buyer client bound to an installed in-process seller platform (test helper).""" + +from __future__ import annotations + +from typing import Any + +from adcp import types as wire +from adcp.types.core import TaskResult, TaskStatus + + +def _plain(value: Any) -> Any: + if hasattr(value, "model_dump"): + return value.model_dump(mode="json", exclude_none=True, by_alias=True) + return value + + +def _result(model: Any, value: Any) -> TaskResult[Any]: + payload = _plain(value) + try: + data = model.model_validate(payload) + except Exception: # noqa: BLE001 - fixture sellers serve minimal buys + data = payload + return TaskResult.model_construct(status=TaskStatus.COMPLETED, data=data, success=True) + + +class InProcessBuyer: + """The tasks ``client.reporting`` reads, dispatched to a platform under one caller.""" + + def __init__(self, platform: Any, context: Any) -> None: + self._platform = platform + self._context = context + self.fail: set[str] = set() + + async def _call(self, task: str, request: Any, model: Any) -> TaskResult[Any]: + if task in self.fail: + return TaskResult.model_construct( + status=TaskStatus.FAILED, data=None, success=False, error=f"{task} unavailable" + ) + params = request.model_dump(mode="json", exclude_none=True) + return _result(model, await getattr(self._platform, task)(params, self._context())) + + async def get_principal(self, request: Any) -> TaskResult[Any]: + return await self._call("get_principal", request, wire.GetPrincipalResponse) + + async def sync_principal(self, request: Any) -> TaskResult[Any]: + return await self._call("sync_principal", request, wire.SyncPrincipalResponse) + + async def list_accounts(self, request: Any) -> TaskResult[Any]: + return await self._call("list_accounts", request, wire.ListAccountsResponse) + + async def get_media_buys(self, request: Any) -> TaskResult[Any]: + return await self._call("get_media_buys", request, wire.GetMediaBuysResponse) + + async def get_reporting_status(self, request: Any) -> TaskResult[Any]: + return await self._call("get_reporting_status", request, wire.GetReportingStatusResponse) + + async def fetch_capabilities(self) -> Any: + return _plain(await self._platform.get_adcp_capabilities({}, self._context())) diff --git a/tests/test_reporting_subscription_reconcile.py b/tests/test_reporting_subscription_reconcile.py new file mode 100644 index 000000000..f3290dbf9 --- /dev/null +++ b/tests/test_reporting_subscription_reconcile.py @@ -0,0 +1,196 @@ +"""The buyer's ``client.reporting.reconcile()`` and ``explain()`` against a managed seller. + +The seller is the SDK's managed path, installed in process; the buyer reads it +only through wire tasks, so roster, grants and buys are all wire-sourced. +""" + +from __future__ import annotations + +from datetime import timedelta + +import pytest + +from adcp.reporting import PrincipalRef +from adcp.reporting.subscriptions import diff_coverage_traces +from adcp.reporting.subscriptions.facade import ReportingClient +from tests.reporting_inprocess import InProcessBuyer +from tests.test_reporting_subscription_managed import ( + ACCOUNT, + AGENCY, + OFFERING, + START, + _ctx, + _World, +) + +pytestmark = pytest.mark.asyncio + + +def _buyer(world: _World) -> tuple[ReportingClient, InProcessBuyer]: + client = InProcessBuyer(world.platform, _ctx) + return ReportingClient(client), client + + +async def _delivered_world() -> _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 + return world + + +async def test_reconcile_derives_the_roster_and_reports_delivered_periods() -> None: + world = await _delivered_world() + reporting, _ = _buyer(world) + report = await reporting.reconcile(now=world.clock()) + assert report.definitive, report.issues + assert report.accounts == (ACCOUNT,) + assert report.summary.owed == report.summary.delivered == 2 + first = report.periods[0] + assert first.to_dict() | {"message": None} == { + "account_id": ACCOUNT, + "subscription_id": "analytics-hourly", + "subscription_version": 1, + "offering_id": OFFERING["offering_id"], + "report_definition_id": first.report_definition_id, + "media_buy_ids": ["media-buy-a1"], + "period": { + "start": "2026-11-01T01:00:00Z", + "end": "2026-11-01T02:00:00Z", + "timezone": "UTC", + }, + "status": "delivered", + "reason": "owed", + "responsible_party": "none", + "message": None, + "expected_at": "2026-11-01T02:10:00Z", + "reporting_obligation_id": first.reporting_obligation_id, + "reporting_revision_id": first.reporting_revision_id, + "finality": first.finality, + } + assert first.message.startswith("Owed: media-buy-a1 under") + assert first.reporting_obligation_id and first.reporting_revision_id + + +async def test_a_missing_obligation_is_the_seller_after_grace_and_pending_before() -> None: + world = _World() + await world.subscribe() + reporting, _ = _buyer(world) + # No worker turn ran: the seller planned nothing. + within = await reporting.reconcile(now=START + timedelta(hours=2, minutes=30)) + assert [line.status for line in within.periods] == ["pending"] + later = await reporting.reconcile(now=START + timedelta(hours=3, minutes=30)) + assert [(line.status, line.responsible_party) for line in later.periods] == [ + ("missing_obligation", "seller"), + ("pending", "none"), + ] + assert later.summary.missing_obligation == 1 and later.summary.pending == 1 + # A zero grace is the caller's explicit choice. + strict = await reporting.reconcile(now=START + timedelta(hours=3), grace=timedelta(0)) + assert [line.status for line in strict.periods] == ["missing_obligation"] * 2 + + +async def test_include_not_owed_lists_the_rest_of_the_flight_with_reasons() -> None: + world = _World() + world.seller.buys[0]["confirmed_at"] = "2026-11-01T02:30:00Z" + await world.subscribe() + world.clock.set(START + timedelta(hours=3, minutes=30)) + await world.service.run_worker() + reporting, _ = _buyer(world) + report = await reporting.reconcile(now=world.clock(), include_not_owed=True) + assert [(line.period["start"], line.status, line.reason) for line in report.periods] == [ + ("2026-11-01T01:00:00Z", "not_owed", "before_confirmation"), + ("2026-11-01T02:00:00Z", "delivered", "owed"), + ] + assert report.summary.not_owed == 1 and report.summary.owed == 1 + + +async def test_since_and_explicit_accounts_narrow_the_report() -> None: + world = await _delivered_world() + reporting, _ = _buyer(world) + report = await reporting.reconcile(now=world.clock(), since="2026-11-01T02:00:00Z") + assert [line.period["start"] for line in report.periods] == ["2026-11-01T02:00:00Z"] + unknown = await reporting.reconcile(now=world.clock(), accounts=["account-unknown"]) + assert not unknown.definitive and unknown.periods == () + assert [issue.code for issue in unknown.issues] == ["SUBSCRIPTION_ACCOUNT_NOT_COVERED"] + + +async def test_a_partial_roster_is_never_a_smaller_success() -> None: + world = await _delivered_world() + reporting, client = _buyer(world) + client.fail.add("list_accounts") + report = await reporting.reconcile(now=world.clock()) + assert not report.definitive + assert "SUBSCRIPTION_ROSTER_UNAVAILABLE" in {issue.code for issue in report.issues} + + +async def test_a_lost_account_reads_from_the_last_facts_and_withholds() -> None: + world = _World() + await world.subscribe() + reporting, _ = _buyer(world) + world.clock.set(START + timedelta(hours=2, minutes=5)) + # The buyer read the account while authorized. + first = await reporting.reconcile(now=world.clock()) + assert first.accounts == (ACCOUNT,) + # The 01:00 period ended at 02:00 and is due 02:10; authority is lost before that. + del world.seller.accounts[ACCOUNT] + await world.service.touch(account_id=ACCOUNT) + await world.service.drain_subscription_triggers() + report = await reporting.reconcile(now=world.clock()) + assert report.accounts == (ACCOUNT,) + assert [(line.status, line.reason) for line in report.periods] == [ + ("withheld", "owed:withheld") + ] + assert report.periods[0].responsible_party == "none" + + +async def test_buyer_and_seller_explain_one_period_identically() -> None: + world = await _delivered_world() + reporting, _ = _buyer(world) + buyer = await reporting.explain( + account_id=ACCOUNT, period="2026-11-01T01:30:00Z", now=world.clock() + ) + seller = await world.service.explain_coverage( + PrincipalRef(None, AGENCY), ACCOUNT, period="2026-11-01T01:30:00Z" + ) + assert [trace.side for trace in buyer] == ["buyer"] + assert (buyer[0].verdict, buyer[0].reason) == ("owed", "owed") + assert buyer[0].inputs_digest == seller[0].inputs_digest + assert diff_coverage_traces(buyer[0], seller[0]) is None + day = await reporting.explain(account_id=ACCOUNT, period="2026-11-01", now=world.clock()) + assert [trace.reason for trace in day][:2] == ["outside_flight", "owed"] + with pytest.raises(KeyError): + await reporting.explain( + account_id=ACCOUNT, period="2026-11-01", subscription_id="missing" + ) + + +async def test_the_cli_reconciles_and_explains_against_an_agent(monkeypatch, capsys) -> None: + import asyncio + import json + + from adcp.reporting import cli + + world = await _delivered_world() + reporting, _ = _buyer(world) + + class Agent: + def __init__(self) -> None: + self.reporting = reporting + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return None + + monkeypatch.setattr(cli, "_agent_client", lambda identifier: Agent()) + loop = asyncio.get_running_loop() + code = await loop.run_in_executor(None, cli.main, ["reconcile", "--agent", "seller", "--json"]) + assert code == 0 + document = json.loads(capsys.readouterr().out) + assert document["summary"]["owed"] == 2 + argv = ["explain", "--agent", "seller", "--account", ACCOUNT, "--period", "2026-11-01"] + assert await loop.run_in_executor(None, cli.main, argv) == 0 + assert "[owed]" in capsys.readouterr().out From 3ce5beb12aaad866d0f871f2cc41e35649392815 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 17:20:06 +0000 Subject: [PATCH 2/5] style(reporting): format the reconcile tests Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/test_reporting_subscription_reconcile.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/test_reporting_subscription_reconcile.py b/tests/test_reporting_subscription_reconcile.py index f3290dbf9..c79125ce1 100644 --- a/tests/test_reporting_subscription_reconcile.py +++ b/tests/test_reporting_subscription_reconcile.py @@ -161,9 +161,7 @@ async def test_buyer_and_seller_explain_one_period_identically() -> None: day = await reporting.explain(account_id=ACCOUNT, period="2026-11-01", now=world.clock()) assert [trace.reason for trace in day][:2] == ["outside_flight", "owed"] with pytest.raises(KeyError): - await reporting.explain( - account_id=ACCOUNT, period="2026-11-01", subscription_id="missing" - ) + await reporting.explain(account_id=ACCOUNT, period="2026-11-01", subscription_id="missing") async def test_the_cli_reconciles_and_explains_against_an_agent(monkeypatch, capsys) -> None: From d476b42a721e1545c781ea57c9f1225100dd812c Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 17:38:22 +0000 Subject: [PATCH 3/5] fix(reporting): report a failed principal read as a reconcile issue reconcile now turns a failed get_principal into a SUBSCRIPTION_PRINCIPAL_UNAVAILABLE issue with an empty, non-definitive report, like every other read. explain, which has no issue channel, still raises. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/reconcile.py | 24 ++++++++++++++++--- .../test_reporting_subscription_reconcile.py | 13 ++++++++++ 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/src/adcp/reporting/subscriptions/reconcile.py b/src/adcp/reporting/subscriptions/reconcile.py index e264f32ff..24a39a74f 100644 --- a/src/adcp/reporting/subscriptions/reconcile.py +++ b/src/adcp/reporting/subscriptions/reconcile.py @@ -447,10 +447,28 @@ def issue( ) -> None: self.issues.append(ReconcileIssue(code, message, account_id, subscription_id)) - async def load(self, *, accounts: Sequence[str] | None) -> list[_Account]: + async def load(self, *, accounts: Sequence[str] | None, strict: bool = False) -> list[_Account]: + """Read the principal, capabilities and roster. + + Every read degrades to an issue, including ``get_principal`` + (``SUBSCRIPTION_PRINCIPAL_UNAVAILABLE``, after which nothing can be + computed and the report is empty and not definitive). ``strict`` + (``explain``, which has no issue channel) raises it instead. + """ + from adcp.reporting.subscriptions.consumer import ReportingSubscriptionError from adcp.reporting.subscriptions.facade import read_principal - principal = await read_principal(self.client) + try: + principal = await read_principal(self.client) + except ReportingSubscriptionError as error: + if strict: + raise + self.issue( + "SUBSCRIPTION_PRINCIPAL_UNAVAILABLE", + f"The subscriptions could not be read ({error}). Nothing was computed; retry " + "once get_principal answers.", + ) + return [] self.subscriptions = principal.subscriptions try: self.seller = await _seller_facts(self.client) @@ -857,7 +875,7 @@ async def explain_buyer_coverage( """The buyer's :class:`CoverageTrace` list. See :meth:`ReportingClient.explain`.""" instant = utc_instant(now, field="now") if now is not None else datetime.now(timezone.utc) context = _Context(client, state or BuyerReportingState()) - records = await context.load(accounts=[account_id]) + records = await context.load(accounts=[account_id], strict=True) record = next((item for item in records if item.account_id == account_id), None) if record is None: record = context.state.accounts.get(account_id) diff --git a/tests/test_reporting_subscription_reconcile.py b/tests/test_reporting_subscription_reconcile.py index c79125ce1..4d7cb105d 100644 --- a/tests/test_reporting_subscription_reconcile.py +++ b/tests/test_reporting_subscription_reconcile.py @@ -125,6 +125,19 @@ async def test_a_partial_roster_is_never_a_smaller_success() -> None: assert "SUBSCRIPTION_ROSTER_UNAVAILABLE" in {issue.code for issue in report.issues} +async def test_a_failed_principal_read_is_an_issue_for_reconcile_and_raises_for_explain() -> None: + from adcp.reporting.subscriptions import ReportingSubscriptionError + + world = await _delivered_world() + reporting, client = _buyer(world) + client.fail.add("get_principal") + report = await reporting.reconcile(now=world.clock()) + assert not report.definitive and report.periods == () + assert [issue.code for issue in report.issues] == ["SUBSCRIPTION_PRINCIPAL_UNAVAILABLE"] + with pytest.raises(ReportingSubscriptionError): + await reporting.explain(account_id=ACCOUNT, period="2026-11-01", now=world.clock()) + + async def test_a_lost_account_reads_from_the_last_facts_and_withholds() -> None: world = _World() await world.subscribe() From e112cc11c103fe2febbd71f13007aa3130da3a1c Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 18:26:46 +0000 Subject: [PATCH 4/5] fix(reporting): report every owed period before an open revocation as withheld A period that ended before the caller lost authority, with no re-grant since, is withheld, never missing or late, whether or not it was due at the loss. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/reconcile.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/adcp/reporting/subscriptions/reconcile.py b/src/adcp/reporting/subscriptions/reconcile.py index 24a39a74f..1e3585261 100644 --- a/src/adcp/reporting/subscriptions/reconcile.py +++ b/src/adcp/reporting/subscriptions/reconcile.py @@ -671,7 +671,8 @@ def _line( if obligation.receipt_owed: responsible = "buyer" receipt = {"status": "owed_by_buyer"} - elif period.withheld: + elif period.withheld or trace_reason == "owed:withheld": + # Withheld while authority is lost: never missing or late. status = "withheld" elif obligation is None: if now >= end + grace: From 9defc4c13f4a32d670d11174c5e2a92cdbf40634 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 18:52:43 +0000 Subject: [PATCH 5/5] fix(reporting): list each flight period once in the not-owed audit flight_candidates enumerated a buy's flight once per zone the offering's periods can take, so include_not_owed could list one period twice and inflate summary.not_owed. Each period is now listed once, as select does. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/explain.py | 7 ++++++- tests/test_reporting_subscription_explain.py | 20 ++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/src/adcp/reporting/subscriptions/explain.py b/src/adcp/reporting/subscriptions/explain.py index 9da50f51d..3185e35f0 100644 --- a/src/adcp/reporting/subscriptions/explain.py +++ b/src/adcp/reporting/subscriptions/explain.py @@ -1328,12 +1328,17 @@ def flight_candidates( listed = sorted({oid for _pkg, ids in buy.product_offering_ids() for oid in ids}) for oid in [o for o in listed if o in self.offerings]: offering = self.offerings[oid] + # One candidate per period, as select() does, however many + # zones the offering's periods can take. + seen: set[tuple[datetime, datetime]] = set() for zone in _zones(context, offering, [low, high]): calendar = self._calendar(offering, zone) for s, e in calendar.periods( starts_from=calendar.floor(low), starts_before=high ): - result.append(_Candidate(buy, offering, calendar, s, e, zone)) + if (s, e) not in seen: + seen.add((s, e)) + result.append(_Candidate(buy, offering, calendar, s, e, zone)) return result def trace(self, candidate: _Candidate) -> CoverageTrace: diff --git a/tests/test_reporting_subscription_explain.py b/tests/test_reporting_subscription_explain.py index b265f07f1..839c162d9 100644 --- a/tests/test_reporting_subscription_explain.py +++ b/tests/test_reporting_subscription_explain.py @@ -491,3 +491,23 @@ def test_an_account_resolved_source_zone_stays_out_of_the_digest() -> None: assert _offering_facts(resolved)["period_timezone"] is None fixed = replace(resolved, period_timezone_policy="fixed") assert _offering_facts(fixed)["period_timezone"] == "America/New_York" + + +def test_flight_candidates_list_each_period_once_across_zones() -> None: + from adcp.reporting.subscriptions import AccountTimezoneChange + from adcp.reporting.subscriptions.explain import CoverageExplainer + + scenario = _scenario() + scenario["timezone_changes"] = [ + AccountTimezoneChange( + datetime(2026, 11, 3, 5, tzinfo=timezone.utc), + "America/Chicago", + previous_timezone="America/New_York", + ) + ] + explainer = CoverageExplainer(**scenario) + keys = [ + (c.buy.media_buy_id, c.offering.offering_id, c.start, c.end) + for c in explainer.flight_candidates() + ] + assert len(keys) == len(set(keys))