fix(reporting): bound each seller read in reconcile; scrub account_requests[]

docs/reporting-subscriptions-buyer.md

@@ -57,7 +57,7 @@ for line in report.periods:
         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.
+`reconcile()` reads everything from the wire on each run. Never compute periods yourself. Each read is bounded by `read_timeout` (default 30 seconds, `None` for no bound), so one unresponsive seller cannot stall the report: a read that does not answer in time fails as `TIMEOUT` like any other failed read.
 - **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.
@@ -95,7 +95,7 @@ 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.
+`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. `read_timeout` bounds each read of each seller, so a hung seller lands in `fleet.issues` (or fails its accounts) and frees its `concurrency` slot instead of holding the run.
 
 From a shell: `python -m adcp reporting reconcile --agent <alias> [--since 2026-11-01] [--json]`.
 

docs/reporting-subscriptions-errors.md

@@ -48,7 +48,7 @@ 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 `"*"`.
+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 `"*"`. A read that did not answer within `read_timeout` has `error.code` `TIMEOUT` and a message naming `read_timeout`.
 
 `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`.
 

src/adcp/decisioning/account_projection.py

@@ -364,6 +364,7 @@ def to_wire_sync_governance_row(row: SyncGovernanceResultRow) -> dict[str, Any]:
 CREDENTIAL_BEARING_METHODS: frozenset[str] = frozenset(
     {
         "list_accounts",
+        "request_account",
         "sync_accounts",
         "sync_governance",
         "create_media_buy",

src/adcp/decisioning/handler.py

@@ -3594,7 +3594,13 @@ async def list_accounts(  # type: ignore[override]
             page = _call_with_optional_ctx(list_requests, params, ctx=resolve_ctx)
             if inspect.isawaitable(page):
                 page = await page
-            served.update(_project_account_requests_page(page))
+            # The rows above were scrubbed before the requests were attached;
+            # a resolved request carries a full Account, so scrub them too.
+            served.update(
+                strip_credentials_from_wire_result(
+                    "list_accounts", _project_account_requests_page(page)
+                )
+            )
         reporting = getattr(self._platform, "_reliable_reporting_service", None)
         if reporting is not None:
             # Managed reporting subscriptions serve the grant journal's
@@ -3623,7 +3629,9 @@ async def request_account(  # type: ignore[override]
         result = _call_with_optional_ctx(request_account, params, ctx=resolve_ctx)
         if inspect.isawaitable(result):
             result = await result
-        return _project_request_account(result)
+        return strip_credentials_from_wire_result(
+            "request_account", _project_request_account(result)
+        )
 
     # ----- Optional-method gate -----
 

src/adcp/reporting/subscriptions/events.py

@@ -45,6 +45,7 @@
 from adcp.reporting.subscriptions._clock import iso_z, utc_instant
 from adcp.reporting.subscriptions.reconcile import (
     DEFAULT_GRACE,
+    DEFAULT_READ_TIMEOUT,
     BuyerReportingState,
     ReconcileReport,
     _reconcile,
@@ -496,11 +497,12 @@ async def poll(
         grace: timedelta = DEFAULT_GRACE,
         baseline: bool = False,
         now: datetime | None = None,
+        read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT,
     ) -> EventBatch:
         """One batch of events since the last acknowledgement; call ``batch.ack()`` after handling.
 
         ``baseline=True`` on a consumer with no snapshot yet records the
-        current state without emitting it.
+        current state without emitting it. ``read_timeout`` bounds each read.
         """
         instant = utc_instant(now, field="now") if now is not None else datetime.now(timezone.utc)
         report, context = await _reconcile(
@@ -513,6 +515,7 @@ async def poll(
             include_not_owed=False,
             now=instant,
             receipts=None,
+            read_timeout=read_timeout,
         )
         current = _snapshot(report, context)
         loaded = await self.store.load(self.agent_id)
@@ -539,6 +542,7 @@ async def events(
         poll_every: timedelta = timedelta(minutes=5),
         stop: asyncio.Event | None = None,
         grace: timedelta = DEFAULT_GRACE,
+        read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT,
     ) -> AsyncIterator[ReportingEvent]:
         """Every event, forever (or until ``stop`` is set).
 
@@ -553,7 +557,7 @@ async def events(
                 # Consume the ring before polling: one that lands during the
                 # poll, the loop body or the ack survives and polls again at once.
                 self._doorbell.clear()
-                batch = await self.poll(grace=grace)
+                batch = await self.poll(grace=grace, read_timeout=read_timeout)
                 for event in batch.events:
                     yield event
                 try:

src/adcp/reporting/subscriptions/facade.py

@@ -49,6 +49,9 @@
 
 #: The default ``grace`` of :meth:`ReportingClient.reconcile`: one hour, never zero.
 DEFAULT_GRACE = timedelta(hours=1)
+#: The default ``read_timeout`` of :meth:`ReportingClient.reconcile`: each
+#: read of one seller is bounded to 30 seconds, as in the TS SDK.
+DEFAULT_READ_TIMEOUT = timedelta(seconds=30)
 
 SubscribeAction = Literal["created", "updated", "unchanged", "reactivated", "deactivated"]
 
@@ -443,6 +446,7 @@ async def reconcile(
         include_not_owed: bool = False,
         receipts: ReceiptPolicy | None = None,
         now: datetime | None = None,
+        read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT,
     ) -> ReconcileReport:
         """What the seller owes every covered account, and whose move each period is.
 
@@ -468,6 +472,11 @@ async def reconcile(
             The default is off: nothing is receipted on read. ``mode="verified"``
             receipts a billing revision only after the SDK verified its
             materialization and your ``commit`` hook returned a reference.
+        :param read_timeout: bounds each read (and the receipts policy's write),
+            so one unresponsive seller cannot stall the report. A read that
+            does not answer in time fails as ``TIMEOUT``, naming
+            ``read_timeout``, the same way any other failed read does. Defaults
+            to 30 seconds; ``None`` for no bound.
 
         ``report.definitive`` is true only when every read completed;
         otherwise ``report.issues`` say what to retry.
@@ -484,6 +493,7 @@ async def reconcile(
             include_not_owed=include_not_owed,
             receipts=receipts,
             now=now,
+            read_timeout=read_timeout,
         )
 
     async def submit_receipts(
@@ -532,6 +542,7 @@ async def explain(
         media_buy_id: str | None = None,
         offering_id: str | None = None,
         now: datetime | None = None,
+        read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT,
     ) -> list[CoverageTrace]:
         """Why each selected period is, or is not, owed: the buyer's ``CoverageTrace`` list.
 
@@ -540,6 +551,9 @@ async def explain(
         ``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`.
+        ``read_timeout`` bounds each read, as in :meth:`reconcile`; a read that
+        does not answer in time raises a ``TIMEOUT``
+        :class:`ReportingSubscriptionError`.
         """
         from adcp.reporting.subscriptions.reconcile import explain_buyer_coverage
 
@@ -552,6 +566,7 @@ async def explain(
             media_buy_id=media_buy_id,
             offering_id=offering_id,
             now=now,
+            read_timeout=read_timeout,
         )
 
     def _event_source(self) -> ReportingEvents:
@@ -567,6 +582,7 @@ async def poll(
         grace: timedelta = DEFAULT_GRACE,
         baseline: bool = False,
         now: datetime | None = None,
+        read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT,
     ) -> EventBatch:
         """One batch of reporting events since the last ``ack``; serverless-safe.
 
@@ -575,22 +591,28 @@ async def poll(
         ``reconcile`` and coverage with the snapshot saved at the last ack, so
         a crash before ``ack`` redelivers the same events with the same ids.
         Pass ``event_store=`` to :class:`ReportingClient` for a durable store.
+        ``read_timeout`` bounds each read, as in :meth:`reconcile`.
         """
-        return await self._event_source().poll(grace=grace, baseline=baseline, now=now)
+        return await self._event_source().poll(
+            grace=grace, baseline=baseline, now=now, read_timeout=read_timeout
+        )
 
     def events(
         self,
         *,
         poll_every: timedelta = timedelta(minutes=5),
         stop: Any = None,
         grace: timedelta = DEFAULT_GRACE,
+        read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT,
     ) -> AsyncIterator[ReportingEvent]:
         """``async for event in client.reporting.events(): ...``, acked per batch.
 
         Polls every ``poll_every``, or at once when :meth:`handle_webhook`
         rings. ``stop`` is an ``asyncio.Event`` that ends the loop.
         """
-        return self._event_source().events(poll_every=poll_every, stop=stop, grace=grace)
+        return self._event_source().events(
+            poll_every=poll_every, stop=stop, grace=grace, read_timeout=read_timeout
+        )
 
     async def handle_webhook(self, verified: Any) -> EventBatch | None:
         """Treat a verified seller webhook as a doorbell.
@@ -1013,6 +1035,7 @@ async def reconcile(
         include_not_owed: bool = False,
         receipts: ReceiptPolicy | None = None,
         now: datetime | None = None,
+        read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT,
     ) -> FleetReconcileReport:
         """:meth:`ReportingClient.reconcile` on every seller, ``concurrency`` at a time.
 
@@ -1021,7 +1044,10 @@ async def reconcile(
         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.
+        :meth:`ReportingClient.reconcile`'s, applied to every seller:
+        ``read_timeout`` bounds each read of each seller, so a seller that
+        stops answering is reported in ``issues`` and frees its slot instead
+        of holding the run.
         """
         from adcp.reporting.subscriptions.reconcile import reconcile_fleet
 
@@ -1036,6 +1062,7 @@ async def reconcile(
             include_not_owed=include_not_owed,
             receipts=receipts,
             now=now,
+            read_timeout=read_timeout,
         )