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 e19024c96..3185e35f0 100644 --- a/src/adcp/reporting/subscriptions/explain.py +++ b/src/adcp/reporting/subscriptions/explain.py @@ -1116,201 +1116,247 @@ 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] + # 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 + ): + if (s, e) not in seen: + seen.add((s, e)) + 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 @@ -1322,17 +1368,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), @@ -1340,44 +1386,90 @@ 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, - ended=candidate.end <= now, + authorization=self._authorization, + backfill=self.backfill, + status_retention_days=self.retention_days, + ended=candidate.end <= self.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, - ) + 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..1e3585261 --- /dev/null +++ b/src/adcp/reporting/subscriptions/reconcile.py @@ -0,0 +1,918 @@ +"""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, 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 + + 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) + 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 or trace_reason == "owed:withheld": + # Withheld while authority is lost: never missing or late. + 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], 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) + 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_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)) diff --git a/tests/test_reporting_subscription_reconcile.py b/tests/test_reporting_subscription_reconcile.py new file mode 100644 index 000000000..4d7cb105d --- /dev/null +++ b/tests/test_reporting_subscription_reconcile.py @@ -0,0 +1,207 @@ +"""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_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() + 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