From 4a1673a10f80409028f29fd09645e47a92e2c7fd Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 00:14:42 +0000 Subject: [PATCH 1/2] feat!: per-account reconcile report and fleet reconcile for reporting subscriptions ReconcileReport.accounts now holds one ReconcileAccount per account, the shape the TS SDK reports: account_id, subscription_ids, a status of reconciled, unauthorized or failed, unauthorized_at and withheld_period_count for a lost account, and an error naming the task, code and message of the read that failed. A principal or roster failure is the entry "*". report.account_ids is the old tuple of ids. The report also carries agent_id and since. ADCPMultiAgentClient.reporting.reconcile fans reconcile out across sellers with bounded concurrency and returns a FleetReconcileReport: by_seller, issues, definitive and the summed summary, matching TS multiClient.reporting.reconcile. A seller whose principal could not be read is a fleet issue with its agent_id. BREAKING CHANGE: ReconcileReport.accounts is a tuple of ReconcileAccount instead of account-id strings, and to_dict serves objects under accounts. Read report.account_ids for the ids. Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 1 + docs/reporting-subscriptions-buyer.md | 19 ++ docs/reporting-subscriptions-errors.md | 4 + src/adcp/client.py | 13 +- src/adcp/reporting/subscriptions/__init__.py | 8 + src/adcp/reporting/subscriptions/events.py | 4 +- src/adcp/reporting/subscriptions/facade.py | 61 +++- src/adcp/reporting/subscriptions/reconcile.py | 278 +++++++++++++++++- .../test_reporting_subscription_reconcile.py | 76 ++++- 9 files changed, 450 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index c79ec9672b..6b878287a8 100644 --- a/README.md +++ b/README.md @@ -366,6 +366,7 @@ platform = reporting.install(MySeller()) # serves get_principal/sync_principal, 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 +fleet = await multi.reporting.reconcile() # every seller of an ADCPMultiAgentClient 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 | ... ``` diff --git a/docs/reporting-subscriptions-buyer.md b/docs/reporting-subscriptions-buyer.md index 14f52baef2..5f70ca20a7 100644 --- a/docs/reporting-subscriptions-buyer.md +++ b/docs/reporting-subscriptions-buyer.md @@ -77,6 +77,25 @@ Each owed period that has ended becomes a `PeriodLine`, with wire-style keys and - `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`. +- `report.accounts` has one `ReconcileAccount` per account, the same shape as the TS SDK's: `account_id`, `subscription_ids` and a `status`. + - `reconciled`: its facts were read and its owed set computed. + - `unauthorized`: your authority ended at `unauthorized_at`; `withheld_period_count` is the seller's count of withheld periods. + - `failed`: `error` (`task`, `code`, `message`) names the read that failed. The account id is `"*"` when the principal or the roster could not be read at all. + + `report.account_ids` is the tuple of ids that `report.accounts` was before. + +### Across sellers + +```python +multi = ADCPMultiAgentClient(agents) +fleet = await multi.reporting.reconcile(since="2026-11-01", concurrency=4) +for agent_id, report in fleet.by_seller.items(): + print(agent_id, report.definitive, report.summary) +for issue in fleet.issues: # sellers that could not be reconciled at all + print(issue.agent_id, issue.code, issue.message) +``` + +`multi.reporting.reconcile()` runs `reconcile()` on every seller, `concurrency` at a time, with the same options. Each seller is read through its own `multi.agent(agent_id).reporting`. `fleet.summary` sums the sellers' summaries. A seller whose principal could not be read is in `fleet.issues` (`SUBSCRIPTION_PRINCIPAL_UNAVAILABLE`, with `agent_id`), not in `by_seller`. `fleet.definitive` is true only when every seller's report is definitive and no seller failed. `agent_ids=` narrows the run. From a shell: `python -m adcp reporting reconcile --agent [--since 2026-11-01] [--json]`. diff --git a/docs/reporting-subscriptions-errors.md b/docs/reporting-subscriptions-errors.md index 2c6e7dfffd..bc512a77de 100644 --- a/docs/reporting-subscriptions-errors.md +++ b/docs/reporting-subscriptions-errors.md @@ -48,6 +48,10 @@ A seller rejection is a value: `AccountRejected`. Its `requires_new_account` is | `SUBSCRIPTION_ACCOUNT_NOT_COVERED` | an explicit `accounts=` id is not readable, or no subscription covers it | check `list_accounts` and the subscription scopes | | `COVERAGE_INPUT_INCOMPLETE`, `COVERAGE_INPUT_INVALID` | an owed set could not be computed, for example a missing account zone | the `detail` names the input | +An account an issue names is also `status: "failed"` in `report.accounts`, with `error.task` set to the read that failed (`get_media_buys`, `get_reporting_status`, `coverage`, ...). A principal or roster failure is the entry `"*"`. + +`multi.reporting.reconcile()` lifts a seller whose principal could not be read, or whose run raised, into `fleet.issues` as `SUBSCRIPTION_PRINCIPAL_UNAVAILABLE` with its `agent_id`. Every other issue stays on that seller's report in `fleet.by_seller`. + ## Events and receipts | Code | Class | `dispatched` | `recovery` | Fix | diff --git a/src/adcp/client.py b/src/adcp/client.py index 74753adcb8..48df5ebaf3 100644 --- a/src/adcp/client.py +++ b/src/adcp/client.py @@ -26,7 +26,7 @@ import httpx from mcp import ClientSession - from adcp.reporting.subscriptions.facade import ReportingClient + from adcp.reporting.subscriptions.facade import FleetReportingClient, ReportingClient from adcp._null_clear import preserve_explicit_nulls from adcp._response_limit import ( @@ -7430,6 +7430,17 @@ def agent_ids(self) -> list[str]: """Get list of agent IDs.""" return list(self.agents.keys()) + @property + def reporting(self) -> FleetReportingClient: + """Fleet-wide reporting reads: ``reconcile`` across every agent. + + See :class:`adcp.reporting.subscriptions.facade.FleetReportingClient`. + Each seller is read through ``multi.agent(agent_id).reporting``. + """ + from adcp.reporting.subscriptions.facade import FleetReportingClient + + return FleetReportingClient(self.agents) + async def close(self) -> None: """Close all agent clients and clean up resources.""" import asyncio diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index 9f4d1af6a5..102c2ccefa 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -122,7 +122,11 @@ ) from adcp.reporting.subscriptions.reconcile import ( BuyerReportingState, + FleetReconcileIssue, + FleetReconcileReport, PeriodLine, + ReconcileAccount, + ReconcileAccountError, ReconcileIssue, ReconcileReport, ReconcileSummary, @@ -168,7 +172,11 @@ "VerifiedWebhook", "BuyerReportingState", "CoverageExplainer", + "FleetReconcileIssue", + "FleetReconcileReport", "PeriodLine", + "ReconcileAccount", + "ReconcileAccountError", "ReconcileIssue", "ReconcileReport", "ReconcileSummary", diff --git a/src/adcp/reporting/subscriptions/events.py b/src/adcp/reporting/subscriptions/events.py index 0dcd3aa3cf..5493ac2f40 100644 --- a/src/adcp/reporting/subscriptions/events.py +++ b/src/adcp/reporting/subscriptions/events.py @@ -244,7 +244,7 @@ def _snapshot(report: ReconcileReport, context: Any) -> dict[str, Any]: exceptions: dict[str, dict[str, Any]] = {} cutovers: dict[str, dict[str, Any]] = {} authorizations: dict[str, dict[str, Any]] = {} - for account_id in report.accounts: + for account_id in report.account_ids: record = context.state.accounts.get(account_id) if record is None: continue @@ -602,7 +602,7 @@ def _merge_partial( previous: Mapping[str, Any], current: dict[str, Any], report: ReconcileReport ) -> dict[str, Any]: unread = {issue.account_id for issue in report.issues if issue.account_id} - everything = not report.accounts or any(issue.account_id is None for issue in report.issues) + everything = not report.account_ids or any(issue.account_id is None for issue in report.issues) merged = dict(current) for section in ("periods", "exceptions", "cutovers"): values = dict(current.get(section) or {}) diff --git a/src/adcp/reporting/subscriptions/facade.py b/src/adcp/reporting/subscriptions/facade.py index 14d090da46..832edf6b51 100644 --- a/src/adcp/reporting/subscriptions/facade.py +++ b/src/adcp/reporting/subscriptions/facade.py @@ -45,7 +45,7 @@ ) from adcp.reporting.subscriptions.explain import CoverageTrace from adcp.reporting.subscriptions.receipts import ReceiptPolicy - from adcp.reporting.subscriptions.reconcile import ReconcileReport + from adcp.reporting.subscriptions.reconcile import FleetReconcileReport, ReconcileReport #: The default ``grace`` of :meth:`ReportingClient.reconcile`: one hour, never zero. DEFAULT_GRACE = timedelta(hours=1) @@ -981,6 +981,64 @@ def fail(detail: str) -> ReportingSubscriptionError: raise fail(f"stores a different configuration at {found}") +class FleetReportingClient: + """``multi_client.reporting``: reconcile every seller of an ``ADCPMultiAgentClient``. + + Each seller is read through its own ``client.reporting``, so the per-seller + cache of last-read facts (what shows an account whose authority was lost) + carries across fleet runs. + """ + + def __init__(self, clients: Mapping[str, Any]) -> None: + self._clients = clients + self._readers: dict[str, Any] = {} + + def _reporting(self, agent_id: str) -> Any: + reader = self._readers.get(agent_id) + if reader is None: + client = self._clients[agent_id] + reader = getattr(client, "reporting", None) or ReportingClient(client) + self._readers[agent_id] = reader + return reader + + async def reconcile( + self, + *, + agent_ids: Sequence[str] | None = None, + concurrency: int = 4, + since: Any = None, + accounts: Sequence[str] | Literal["covered"] = "covered", + subscription_id: str | None = None, + grace: timedelta = DEFAULT_GRACE, + include_not_owed: bool = False, + receipts: ReceiptPolicy | None = None, + now: datetime | None = None, + ) -> FleetReconcileReport: + """:meth:`ReportingClient.reconcile` on every seller, ``concurrency`` at a time. + + Returns a :class:`~adcp.reporting.subscriptions.reconcile.FleetReconcileReport`: + ``by_seller`` (agent id to report), ``issues`` (sellers that could not + be reconciled at all, each with ``agent_id``), ``definitive`` and the + summed ``summary``. ``agent_ids`` narrows the run; an unknown id is + refused before anything is sent. The other options are + :meth:`ReportingClient.reconcile`'s, applied to every seller. + """ + from adcp.reporting.subscriptions.reconcile import reconcile_fleet + + return await reconcile_fleet( + {key: self._reporting(key) for key in self._clients}, + agent_ids=agent_ids, + concurrency=concurrency, + since=since, + accounts=accounts, + subscription_id=subscription_id, + grace=grace, + include_not_owed=include_not_owed, + receipts=receipts, + now=now, + ) + + @dataclass class _Plan: configuration: dict[str, Any] | None @@ -1050,6 +1108,7 @@ def _mismatches( __all__ = [ + "FleetReportingClient", "PreviewResult", "ReportingClient", "SubscribeResult", diff --git a/src/adcp/reporting/subscriptions/reconcile.py b/src/adcp/reporting/subscriptions/reconcile.py index b97c923a58..7ca4082d8e 100644 --- a/src/adcp/reporting/subscriptions/reconcile.py +++ b/src/adcp/reporting/subscriptions/reconcile.py @@ -32,6 +32,7 @@ from __future__ import annotations +import asyncio import logging from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass, field @@ -161,6 +162,55 @@ class ReconcileIssue: subscription_id: str | None = None +@dataclass(frozen=True) +class ReconcileAccountError: + """What could not be read for one account: the task, a code and the detail.""" + + task: str + code: str + message: str + + def to_dict(self) -> dict[str, str]: + return {"task": self.task, "code": self.code, "message": self.message} + + +AccountStatus = Literal["reconciled", "unauthorized", "failed"] + + +@dataclass(frozen=True) +class ReconcileAccount: + """One account the run reconciled, or could not; the same shape as the TS SDK's. + + * ``reconciled``: its facts were read and its owed set computed. + * ``unauthorized``: the caller's authority ended at ``unauthorized_at``; + ``withheld_period_count`` is the seller's count of withheld periods. + * ``failed``: ``error`` says which read failed. Its lines, if any, come + from the last facts read for it. ``account_id`` is ``"*"`` when the + principal or the roster could not be read at all. + """ + + account_id: str + subscription_ids: tuple[str, ...] = () + status: AccountStatus = "reconciled" + unauthorized_at: str | None = None + withheld_period_count: int | None = None + error: ReconcileAccountError | None = None + + def to_dict(self) -> dict[str, Any]: + result: dict[str, Any] = { + "account_id": self.account_id, + "subscription_ids": list(self.subscription_ids), + "status": self.status, + } + if self.unauthorized_at is not None: + result["unauthorized_at"] = self.unauthorized_at + if self.withheld_period_count is not None: + result["withheld_period_count"] = self.withheld_period_count + if self.error is not None: + result["error"] = self.error.to_dict() + return result + + @dataclass(frozen=True) class ReconcileReport: """The result of ``client.reporting.reconcile()``. @@ -176,19 +226,36 @@ class ReconcileReport: summary: ReconcileSummary periods: tuple[PeriodLine, ...] issues: tuple[ReconcileIssue, ...] = () - accounts: tuple[str, ...] = () + #: One entry per account, with its status (``reconciled``, ``unauthorized`` + #: or ``failed``). Before 3.3 this was a tuple of account ids; that shape is + #: :attr:`account_ids`. + accounts: tuple[ReconcileAccount, ...] = () #: What the receipts policy did; ``None`` with ``mode="off"``. receipts: ReceiptReport | None = None + #: The seller's agent id, when the client has one. + agent_id: str | None = None + #: The window start the run used; ``None`` when no ``since`` was given. + since: str | None = None + + @property + def account_ids(self) -> tuple[str, ...]: + """The ids of :attr:`accounts`, the shape ``accounts`` had before.""" + return tuple(item.account_id for item in self.accounts if item.account_id != "*") def to_dict(self) -> dict[str, Any]: - return { + result: dict[str, Any] = { "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), + "accounts": [item.to_dict() for item in self.accounts], } + if self.agent_id is not None: + result["agent_id"] = self.agent_id + if self.since is not None: + result["since"] = self.since + return result # -- wire reads ---------------------------------------------------------------------- @@ -212,9 +279,22 @@ def _data(result: Any, task: str) -> Any: class _UnreadableError(Exception): - def __init__(self, task: str, detail: str) -> None: + def __init__(self, task: str, detail: str, code: str = "TASK_FAILED") -> None: super().__init__(f"{task} failed: {detail}") self.task = task + self.code = code + + +def _failure(error: BaseException, task: str) -> ReconcileAccountError: + """The ``error`` of a ``failed`` account: the task that failed, its code and detail.""" + if isinstance(error, _UnreadableError): + return ReconcileAccountError(error.task, error.code, str(error)) + if isinstance(error, CoverageInputError): + return ReconcileAccountError("coverage", error.code, str(error)) + code = getattr(error, "code", None) + return ReconcileAccountError( + task, str(getattr(code, "value", code) or "TASK_FAILED"), str(error) or type(error).__name__ + ) def _instant(value: Any) -> datetime | None: @@ -310,7 +390,9 @@ async def _roster(client: Any) -> list[dict[str, Any]]: 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") + raise _UnreadableError( + "list_accounts", f"did not finish paging within {_MAX_PAGES} pages", "LIMIT_EXCEEDED" + ) async def _buys(client: Any, account_id: str) -> list[dict[str, Any]]: @@ -336,7 +418,9 @@ async def _buys(client: Any, account_id: str) -> list[dict[str, Any]]: 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") + raise _UnreadableError( + "get_media_buys", f"did not finish paging within {_MAX_PAGES} pages", "LIMIT_EXCEEDED" + ) async def _coverage(client: Any, account_id: str) -> dict[str, dict[str, Any]]: @@ -447,6 +531,17 @@ def __init__(self, client: Any, state: BuyerReportingState) -> None: self.subscriptions: tuple[Any, ...] = () self.seller: _Seller | None = None self.roster_complete = False + #: account_id -> the latest ``unauthorized_at`` in ``accounts_unauthorized[]`` + self.unauthorized: dict[str, datetime] = {} + #: account_id -> the seller's withheld-period count, summed over subscriptions + self.withheld: dict[str, int] = {} + #: account_id -> the first read that failed for it + self.failures: dict[str, ReconcileAccountError] = {} + #: the principal or roster read that failed, which no account can name + self.roster_failure: ReconcileAccountError | None = None + + def fail(self, account_id: str, error: BaseException, task: str) -> None: + self.failures.setdefault(account_id, _failure(error, task)) def issue( self, @@ -474,6 +569,7 @@ async def load(self, *, accounts: Sequence[str] | None, strict: bool = False) -> except ReportingSubscriptionError as error: if strict: raise + self.roster_failure = _failure(error, "get_principal") self.issue( "SUBSCRIPTION_PRINCIPAL_UNAVAILABLE", f"The subscriptions could not be read ({error}). Nothing was computed; retry " @@ -496,6 +592,8 @@ async def load(self, *, accounts: Sequence[str] | None, strict: bool = False) -> if at is not None: previous = unauthorized.get(entry.account_id) unauthorized[entry.account_id] = max(at, previous) if previous else at + count = getattr(entry, "withheld_period_count", None) or 0 + self.withheld[entry.account_id] = self.withheld.get(entry.account_id, 0) + count listed = getattr(state, "accounts_unauthorized_count", None) inline = len(getattr(state, "accounts_unauthorized", None) or ()) if listed is not None and listed > inline: @@ -510,6 +608,7 @@ async def load(self, *, accounts: Sequence[str] | None, strict: bool = False) -> self.roster_complete = True except Exception as error: # noqa: BLE001 roster = [] + self.roster_failure = _failure(error, "list_accounts") self.issue( "SUBSCRIPTION_ROSTER_UNAVAILABLE", f"The account roster could not be read completely ({error}); a partial roster " @@ -528,6 +627,7 @@ async def load(self, *, accounts: Sequence[str] | None, strict: bool = False) -> authorization = raw.get("authorization") or {} record.authorized_at = _instant(authorization.get("authorized_at")) record.regranted_at = _instant(authorization.get("regranted_at")) + self.unauthorized = unauthorized for account_id, lost in unauthorized.items(): record = self.state.account(account_id) if account_id not in seen: @@ -567,6 +667,7 @@ async def refresh(self, record: _Account) -> bool: 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.fail(record.account_id, error, "get_media_buys") self.issue( "SUBSCRIPTION_ACCOUNT_UNREADABLE", f"Account {record.account_id!r} could not be read ({error}). The last facts " @@ -801,6 +902,9 @@ async def _reconcile( context = _Context(client, state or BuyerReportingState()) records = await context.load(accounts=None if accounts == "covered" else list(accounts)) lines: list[PeriodLine] = [] + entries: list[ReconcileAccount] = [] + if context.roster_failure is not None: + entries.append(ReconcileAccount("*", (), "failed", error=context.roster_failure)) complete = context.roster_complete for record in records: readable = await context.refresh(record) @@ -818,12 +922,14 @@ async def _reconcile( complete = complete and walked except Exception as error: # noqa: BLE001 complete = False + context.fail(record.account_id, error, "get_reporting_status") 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, ) + subscription_ids: list[str] = [] for state_item in context.subscriptions: sub_id = str(state_item.subscription_id) if subscription_id is not None and sub_id != subscription_id: @@ -833,10 +939,12 @@ async def _reconcile( for entry in getattr(state_item, "accounts_unauthorized", None) or () ): continue + subscription_ids.append(sub_id) try: explainer = context.explainer(state_item, record, instant) except CoverageInputError as error: complete = False + context.fail(record.account_id, error, "coverage") context.issue( error.code, f"{sub_id} on {record.account_id!r}: {error}", @@ -868,6 +976,7 @@ async def _reconcile( if trace.verdict == "owed": continue lines.append(_not_owed_line(trace)) + entries.append(_account_entry(context, record, subscription_ids)) lines.sort( key=lambda line: ( line.account_id.encode(), @@ -900,12 +1009,39 @@ async def _reconcile( summary=ReconcileSummary(**counts), periods=tuple(lines), issues=tuple(context.issues), - accounts=tuple(record.account_id for record in records), + accounts=tuple(sorted(entries, key=lambda entry: entry.account_id.encode())), receipts=receipt_report if policy is not None and policy.mode != "off" else None, + agent_id=_agent_id(client), + since=None if lower is None else iso_z(lower), ) return report, context +def _agent_id(client: Any) -> str | None: + config = getattr(client, "agent_config", None) + identifier = getattr(config, "id", None) + return None if identifier is None else str(identifier) + + +def _account_entry( + context: _Context, record: _Account, subscription_ids: Sequence[str] +) -> ReconcileAccount: + ids = tuple(sorted(set(subscription_ids), key=str.encode)) + failure = context.failures.get(record.account_id) + if failure is not None: + return ReconcileAccount(record.account_id, ids, "failed", error=failure) + lost = context.unauthorized.get(record.account_id) + if not record.listed and lost is not None: + return ReconcileAccount( + record.account_id, + ids, + "unauthorized", + unauthorized_at=iso_z(lost), + withheld_period_count=context.withheld.get(record.account_id, 0), + ) + return ReconcileAccount(record.account_id, ids, "reconciled") + + def _not_owed_line(trace: CoverageTrace) -> PeriodLine: subject = trace.subject return PeriodLine( @@ -923,6 +1059,129 @@ def _not_owed_line(trace: CoverageTrace) -> PeriodLine: ) +# -- fleet ------------------------------------------------------------------------------ + + +@dataclass(frozen=True) +class FleetReconcileIssue: + """A seller the fleet run could not reconcile at all, with its ``agent_id``.""" + + agent_id: str + code: str + message: str + account_id: str | None = None + subscription_id: str | None = None + + def to_dict(self) -> dict[str, Any]: + return dict(vars(self)) + + +@dataclass(frozen=True) +class FleetReconcileReport: + """``multi_client.reporting.reconcile()``: one report per seller and the summed summary. + + ``by_seller`` maps an agent id to its :class:`ReconcileReport`, whose own + ``issues`` and ``accounts`` say what that run could not read. ``issues`` + names the sellers that could not be reconciled at all (their principal + could not be read, or the run raised); they are not in ``by_seller`` and + their periods are not in ``summary``. ``definitive`` is true only when + every seller's report is definitive and no seller failed. + """ + + by_seller: Mapping[str, ReconcileReport] + issues: tuple[FleetReconcileIssue, ...] + definitive: bool + summary: ReconcileSummary + + def to_dict(self) -> dict[str, Any]: + return { + "by_seller": {key: report.to_dict() for key, report in self.by_seller.items()}, + "issues": [issue.to_dict() for issue in self.issues], + "definitive": self.definitive, + "summary": self.summary.to_dict(), + } + + +async def reconcile_fleet( + clients: Mapping[str, Any], + *, + agent_ids: Sequence[str] | None = None, + concurrency: int = 4, + since: Any = None, + accounts: Sequence[str] | Literal["covered"] = "covered", + subscription_id: str | None = None, + grace: timedelta = DEFAULT_GRACE, + include_not_owed: bool = False, + receipts: ReceiptPolicy | None = None, + now: datetime | None = None, +) -> FleetReconcileReport: + """Reconcile each seller with bounded concurrency. See :meth:`FleetReportingClient.reconcile`. + + ``clients`` maps an agent id to an object with a ``reconcile`` coroutine + taking :func:`reconcile_subscriptions`' keyword options (a + :class:`~adcp.reporting.subscriptions.facade.ReportingClient`). + """ + from adcp.reporting.subscriptions.facade import _invalid + + if type(concurrency) is not int or concurrency < 1: + raise _invalid("concurrency must be a positive integer. Pass concurrency=4 or more.") + if agent_ids is not None: + unknown = sorted(set(agent_ids) - set(clients)) + if unknown: + raise _invalid(f"Unknown agent ids {unknown}. Pass ids from multi_client.agent_ids.") + selected = [key for key in clients if agent_ids is None or key in agent_ids] + options: dict[str, Any] = { + "since": since, + "accounts": accounts, + "subscription_id": subscription_id, + "grace": grace, + "include_not_owed": include_not_owed, + "receipts": receipts, + "now": now, + } + reports: dict[str, ReconcileReport] = {} + issues: list[FleetReconcileIssue] = [] + gate = asyncio.Semaphore(concurrency) + + async def one(agent_id: str) -> None: + async with gate: + try: + report = await clients[agent_id].reconcile(**options) + except Exception as error: # noqa: BLE001 - one seller never fails the fleet + code = getattr(error, "code", None) or "TASK_FAILED" + issues.append( + FleetReconcileIssue( + agent_id, + "SUBSCRIPTION_PRINCIPAL_UNAVAILABLE", + f"{error} ({getattr(code, 'value', code)})", + ) + ) + return + unread = next( + (item for item in report.issues if item.code == "SUBSCRIPTION_PRINCIPAL_UNAVAILABLE"), + None, + ) + if unread is not None: + # Nothing could be computed for this seller: a fleet issue, as in the TS SDK. + issues.append(FleetReconcileIssue(agent_id, unread.code, unread.message)) + return + reports[agent_id] = report + + await asyncio.gather(*(one(agent_id) for agent_id in selected)) + counts = {name: 0 for name in ReconcileSummary.__dataclass_fields__} + for report in reports.values(): + for name in counts: + counts[name] += getattr(report.summary, name) + by_seller = {key: reports[key] for key in selected if key in reports} + ordered = sorted(issues, key=lambda issue: selected.index(issue.agent_id)) + return FleetReconcileReport( + by_seller=by_seller, + issues=tuple(ordered), + definitive=not ordered and all(report.definitive for report in by_seller.values()), + summary=ReconcileSummary(**counts), + ) + + async def explain_buyer_coverage( client: Any, *, @@ -968,12 +1227,17 @@ async def explain_buyer_coverage( __all__ = [ "DEFAULT_GRACE", "BuyerReportingState", + "FleetReconcileIssue", + "FleetReconcileReport", "PeriodLine", "PeriodStatus", + "ReconcileAccount", + "ReconcileAccountError", "ReconcileIssue", "ReconcileReport", "ReconcileSummary", "ResponsibleParty", "explain_buyer_coverage", + "reconcile_fleet", "reconcile_subscriptions", ] diff --git a/tests/test_reporting_subscription_reconcile.py b/tests/test_reporting_subscription_reconcile.py index 4d7cb105d2..2aaea4f1f7 100644 --- a/tests/test_reporting_subscription_reconcile.py +++ b/tests/test_reporting_subscription_reconcile.py @@ -45,7 +45,11 @@ async def test_reconcile_derives_the_roster_and_reports_delivered_periods() -> N reporting, _ = _buyer(world) report = await reporting.reconcile(now=world.clock()) assert report.definitive, report.issues - assert report.accounts == (ACCOUNT,) + assert [item.to_dict() for item in report.accounts] == [ + {"account_id": ACCOUNT, "subscription_ids": ["analytics-hourly"], "status": "reconciled"} + ] + assert report.account_ids == (ACCOUNT,) + assert report.to_dict()["accounts"][0]["status"] == "reconciled" assert report.summary.owed == report.summary.delivered == 2 first = report.periods[0] assert first.to_dict() | {"message": None} == { @@ -123,6 +127,23 @@ async def test_a_partial_roster_is_never_a_smaller_success() -> None: report = await reporting.reconcile(now=world.clock()) assert not report.definitive assert "SUBSCRIPTION_ROSTER_UNAVAILABLE" in {issue.code for issue in report.issues} + roster = report.accounts[0] + assert (roster.account_id, roster.status) == ("*", "failed") + assert roster.error is not None and roster.error.task == "list_accounts" + assert "*" not in report.account_ids + + +async def test_an_account_whose_reads_fail_is_failed_with_the_task() -> None: + world = await _delivered_world() + reporting, client = _buyer(world) + client.fail.add("get_reporting_status") + report = await reporting.reconcile(now=world.clock()) + assert not report.definitive + (entry,) = report.accounts + assert (entry.account_id, entry.status) == (ACCOUNT, "failed") + assert entry.error is not None + assert (entry.error.task, entry.error.code) == ("get_reporting_status", "TASK_FAILED") + assert entry.to_dict()["error"]["task"] == "get_reporting_status" async def test_a_failed_principal_read_is_an_issue_for_reconcile_and_raises_for_explain() -> None: @@ -134,6 +155,7 @@ async def test_a_failed_principal_read_is_an_issue_for_reconcile_and_raises_for_ 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"] + assert [(item.account_id, item.status) for item in report.accounts] == [("*", "failed")] with pytest.raises(ReportingSubscriptionError): await reporting.explain(account_id=ACCOUNT, period="2026-11-01", now=world.clock()) @@ -145,13 +167,17 @@ async def test_a_lost_account_reads_from_the_last_facts_and_withholds() -> None: 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,) + assert first.account_ids == (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,) + (entry,) = report.accounts + assert (entry.account_id, entry.status) == (ACCOUNT, "unauthorized") + assert entry.unauthorized_at == "2026-11-01T02:05:00Z" + assert entry.withheld_period_count is not None and entry.error is None + assert entry.subscription_ids == ("analytics-hourly",) assert [(line.status, line.reason) for line in report.periods] == [ ("withheld", "owed:withheld") ] @@ -205,3 +231,47 @@ async def __aexit__(self, *exc): 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 + + +async def test_the_fleet_reconciles_each_seller_and_names_the_unreadable_ones() -> None: + from adcp.reporting.subscriptions import ReportingSubscriptionError + from adcp.reporting.subscriptions.facade import FleetReportingClient + + world = await _delivered_world() + healthy = InProcessBuyer(world.platform, _ctx) + broken = InProcessBuyer(world.platform, _ctx) + broken.fail.add("get_principal") + partial = InProcessBuyer(world.platform, _ctx) + partial.fail.add("get_reporting_status") + fleet = FleetReportingClient({"a": healthy, "b": broken, "c": partial}) + report = await fleet.reconcile(now=world.clock(), concurrency=2) + assert sorted(report.by_seller) == ["a", "c"] + assert report.by_seller["a"].definitive and not report.by_seller["c"].definitive + assert [(issue.agent_id, issue.code) for issue in report.issues] == [ + ("b", "SUBSCRIPTION_PRINCIPAL_UNAVAILABLE") + ] + assert not report.definitive + assert report.summary.owed == 4 and report.summary.delivered == 2 + assert report.to_dict()["issues"][0]["agent_id"] == "b" + only = await fleet.reconcile(now=world.clock(), agent_ids=["a"]) + assert list(only.by_seller) == ["a"] and only.definitive and only.issues == () + with pytest.raises(ReportingSubscriptionError, match="concurrency"): + await fleet.reconcile(concurrency=0) + with pytest.raises(ReportingSubscriptionError, match="Unknown agent ids"): + await fleet.reconcile(agent_ids=["z"]) + + +async def test_the_multi_agent_client_exposes_the_fleet() -> None: + from adcp.client import ADCPMultiAgentClient + from adcp.reporting.subscriptions.facade import FleetReportingClient + from adcp.types.core import AgentConfig, Protocol + + multi = ADCPMultiAgentClient( + [ + AgentConfig(id="s1", agent_uri="https://s1.example/mcp", protocol=Protocol.MCP), + AgentConfig(id="s2", agent_uri="https://s2.example/mcp", protocol=Protocol.MCP), + ] + ) + fleet = multi.reporting + assert isinstance(fleet, FleetReportingClient) + assert fleet._reporting("s1") is multi.agent("s1").reporting From d7b6e3bb84bc8dc8817226d9615ad333cad90144 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 00:31:52 +0000 Subject: [PATCH 2/2] fix: list only the accounts the filtered subscription covers in reconcile Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/reconcile.py | 4 +++- tests/test_reporting_subscription_reconcile.py | 2 ++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/src/adcp/reporting/subscriptions/reconcile.py b/src/adcp/reporting/subscriptions/reconcile.py index 7ca4082d8e..7e3208b2e1 100644 --- a/src/adcp/reporting/subscriptions/reconcile.py +++ b/src/adcp/reporting/subscriptions/reconcile.py @@ -976,7 +976,9 @@ async def _reconcile( if trace.verdict == "owed": continue lines.append(_not_owed_line(trace)) - entries.append(_account_entry(context, record, subscription_ids)) + if subscription_ids or subscription_id is None or record.account_id in context.failures: + # With subscription_id=, an account only other subscriptions cover is not listed. + entries.append(_account_entry(context, record, subscription_ids)) lines.sort( key=lambda line: ( line.account_id.encode(), diff --git a/tests/test_reporting_subscription_reconcile.py b/tests/test_reporting_subscription_reconcile.py index 2aaea4f1f7..88e367c51e 100644 --- a/tests/test_reporting_subscription_reconcile.py +++ b/tests/test_reporting_subscription_reconcile.py @@ -115,6 +115,8 @@ async def test_since_and_explicit_accounts_narrow_the_report() -> None: 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"] + other = await reporting.reconcile(now=world.clock(), subscription_id="not-this-one") + assert other.accounts == () and other.periods == () 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"]