feat(reporting): add the buyer receipts policy to reconcile

README.md

@@ -374,6 +374,7 @@ outcome = await client.accounts.ensure(key, billing="operator")  # AccountReady
 - **Grants.** The SDK is the only source of `authorized_at`. End a relationship with `wind_down_grant()` instead of revoking.
 - **Explain.** Seller `reporting.explain_coverage(...)` and buyer `client.reporting.explain(...)` return the same `CoverageTrace`; `python -m adcp reporting diff` compares them.
 - **Events.** `client.reporting.poll()` / `events()` derive events from saved state with deterministic ids; `handle_webhook()` treats a verified webhook as a doorbell.
+- **Receipts.** Off by default; `reconcile(receipts=ReceiptPolicy(mode="verified", commit=...))` receipts a billing revision only after the SDK verified it and your commit returned a reference.
 - **Coming in later betas:** 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)

docs/reporting-subscriptions-buyer.md

@@ -131,4 +131,27 @@ await client.reporting.handle_webhook(VerifiedWebhook.from_outcome(outcome))
 
 ## Receipts for billing feeds
 
-The receipts policy arrives in a later beta. Submit receipts with `sync_reporting_receipts` after your warehouse commits the data.
+A receipt states that you verified and committed the data, so the SDK never sends one on read. The policy is off by default.
+
+```python
+async def commit(revision: VerifiedRevision):
+    job = await warehouse.load(revision.resource)    # your ingest
+    return CommitResult(commit_ref=job.id)          # or "defer", or Reject(["ROW_COUNT_MISMATCH"])
+
+report = await client.reporting.reconcile(
+    receipts=ReceiptPolicy(mode="verified", commit=commit, resource_reader=reader))
+report.receipts.submitted, report.receipts.deferred, report.receipts.rejected
+```
+
+With `mode="verified"`, the SDK handles each `consumer_receipt` obligation whose current revision is materialized and not yet receipted:
+1. It verifies the materialization itself: the manifest through `resource_reader` (the built-in inspector), or your `inspect` adapter. This gives the observed row count, control totals and digest.
+2. If verification fails, `commit` is not called:
+   - A materialization the built-in inspector refuses (a digest, row-count or object mismatch), or cannot read, goes to `report.receipts.failed` with the code, and nothing is receipted for it.
+   - An `inspect` adapter that returns a mismatched observation gets a `rejected` receipt with the SDK's codes.
+3. Otherwise `commit` runs, and the `accepted` receipt carries your `consumer_commit_ref`. `commit` runs at least once per revision. A `commit` that raises is listed in `report.receipts.failed` as `COMMIT_FAILED`, and a receipt write that fails after it is retried, so the next reconcile calls `commit` again. Make it idempotent on `revision.resource`, for example by keying the load on the materialization id.
+
+Other rules:
+- **Mode off.** Billing periods read `receipt: {"status": "owed_by_buyer"}` and `responsible_party: "buyer"`, and the client logs a warning once.
+- **Adjustments.** They are offered to `adjustments(adjustment)` and deferred by default. Billing never auto-accepts. Accept or reject an adjustment with its raw evidence through `client.reporting.submit_receipts(account_id, adjustment_receipts=[...])`, the typed manual path over `sync_reporting_receipts`.
+- **Once per materialization.** Receipt ids are stable per obligation, revision and materialization. A materialization already receipted, accepted or rejected, is not receipted again.
+- **Ownership.** Receipts are submitted under the principal that owns the subscription: the client you call `reconcile` on.

src/adcp/reporting/subscriptions/__init__.py

@@ -113,6 +113,13 @@
     SubscriptionPlanResult,
     subscription_obligation_id,
 )
+from adcp.reporting.subscriptions.receipts import (
+    CommitResult,
+    ReceiptPolicy,
+    ReceiptReport,
+    Reject,
+    VerifiedRevision,
+)
 from adcp.reporting.subscriptions.reconcile import (
     BuyerReportingState,
     PeriodLine,
@@ -148,6 +155,11 @@
 from adcp.server.reporting_grants import REPORTING_WIND_DOWN_TASKS, wind_down_grant
 
 __all__ = [
+    "CommitResult",
+    "ReceiptPolicy",
+    "ReceiptReport",
+    "Reject",
+    "VerifiedRevision",
     "EventBatch",
     "InMemoryReportingEventStore",
     "ReportingEvent",

src/adcp/reporting/subscriptions/events.py

@@ -512,6 +512,7 @@ async def poll(
             grace=grace,
             include_not_owed=False,
             now=instant,
+            receipts=None,
         )
         current = _snapshot(report, context)
         loaded = await self.store.load(self.agent_id)

src/adcp/reporting/subscriptions/facade.py

@@ -44,6 +44,7 @@
         ReportingEventStore,
     )
     from adcp.reporting.subscriptions.explain import CoverageTrace
+    from adcp.reporting.subscriptions.receipts import ReceiptPolicy
     from adcp.reporting.subscriptions.reconcile import ReconcileReport
 
 #: The default ``grace`` of :meth:`ReportingClient.reconcile`: one hour, never zero.
@@ -414,6 +415,7 @@ async def reconcile(
         subscription_id: str | None = None,
         grace: timedelta = DEFAULT_GRACE,
         include_not_owed: bool = False,
+        receipts: ReceiptPolicy | None = None,
         now: datetime | None = None,
     ) -> ReconcileReport:
         """What the seller owes every covered account, and whose move each period is.
@@ -436,6 +438,10 @@ async def reconcile(
             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).
+        :param receipts: a :class:`~adcp.reporting.subscriptions.ReceiptPolicy`.
+            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.
 
         ``report.definitive`` is true only when every read completed;
         otherwise ``report.issues`` say what to retry.
@@ -450,9 +456,47 @@ async def reconcile(
             subscription_id=subscription_id,
             grace=grace,
             include_not_owed=include_not_owed,
+            receipts=receipts,
             now=now,
         )
 
+    async def submit_receipts(
+        self,
+        account_id: str,
+        *,
+        receipts: Sequence[Any] = (),
+        adjustment_receipts: Sequence[Any] = (),
+        idempotency_key: str | None = None,
+    ) -> list[Any]:
+        """Submit receipts by hand (``sync_reporting_receipts``); returns the item results.
+
+        The manual path, for receipts and adjustment receipts your own
+        pipeline built. Raises :class:`ReportingSubscriptionError` when the
+        task fails; item-level failures are in the returned results.
+        """
+        from adcp.types import SyncReportingReceiptsRequest
+
+        if not receipts and not adjustment_receipts:
+            raise _invalid("Nothing to submit. Pass receipts or adjustment_receipts.")
+        payload: dict[str, Any] = {
+            "account": {"account_id": account_id},
+            "idempotency_key": idempotency_key or f"rrcpt-{uuid4()}",
+        }
+        if receipts:
+            payload["receipts"] = [_dump(item) for item in receipts]
+        if adjustment_receipts:
+            payload["adjustment_receipts"] = [_dump(item) for item in adjustment_receipts]
+        result = await self._client.sync_reporting_receipts(
+            SyncReportingReceiptsRequest.model_validate(payload)
+        )
+        data = getattr(result, "data", None)
+        if not getattr(result, "success", False) or data is None:
+            raise ReportingSubscriptionError(
+                f"sync_reporting_receipts failed: {getattr(result, 'error', None) or 'no response'}"
+                ". Retry with the same idempotency_key."
+            )
+        return list(getattr(data, "results", None) or [])
+
     async def explain(
         self,
         *,