feat!: per-account reconcile report and fleet reconcile for reporting subscriptions

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 | ...
 ```

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 <alias> [--since 2026-11-01] [--json]`.
 

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 |

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

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",

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 {})

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",