diff --git a/README.md b/README.md
index 1176949c1..742527fe9 100644
--- a/README.md
+++ b/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)
diff --git a/docs/reporting-subscriptions-buyer.md b/docs/reporting-subscriptions-buyer.md
index 588ae8543..14f52baef 100644
--- a/docs/reporting-subscriptions-buyer.md
+++ b/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.
diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py
index 7a405a0a8..9f4d1af6a 100644
--- a/src/adcp/reporting/subscriptions/__init__.py
+++ b/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",
diff --git a/src/adcp/reporting/subscriptions/events.py b/src/adcp/reporting/subscriptions/events.py
index 0421fa79b..0dcd3aa3c 100644
--- a/src/adcp/reporting/subscriptions/events.py
+++ b/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)
diff --git a/src/adcp/reporting/subscriptions/facade.py b/src/adcp/reporting/subscriptions/facade.py
index 616f58c07..63adedbfa 100644
--- a/src/adcp/reporting/subscriptions/facade.py
+++ b/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,
         *,
diff --git a/src/adcp/reporting/subscriptions/receipts.py b/src/adcp/reporting/subscriptions/receipts.py
new file mode 100644
index 000000000..bfb4c4d7c
--- /dev/null
+++ b/src/adcp/reporting/subscriptions/receipts.py
@@ -0,0 +1,308 @@
+"""The buyer's receipts policy for ``client.reporting.reconcile`` (AdCP 3.3, experimental).
+
+A receipt says the buyer verified and committed the data, so it is never sent
+on read. With ``ReceiptPolicy(mode="verified", commit=...)``, for each
+``consumer_receipt`` obligation whose current revision is materialized and not
+yet receipted:
+
+1. the SDK verifies the materialization itself: the manifest or native commit
+   through ``resource_reader`` (the built-in manifest inspector) or your
+   ``inspect`` adapter, giving the observed row count, control totals and
+   digest;
+2. a materialization the verifier cannot read, or whose content the
+   built-in inspector refuses (a digest, row-count or object mismatch), is
+   listed in ``report.receipts.failed`` with the code and receipted nothing;
+   an ``inspect`` adapter that returns a mismatched observation gets a
+   ``rejected`` receipt with the SDK's codes (``ROW_COUNT_MISMATCH`` and
+   friends), and ``commit`` is not called;
+3. otherwise your ``commit(revision)`` hook runs. It runs **at least once**
+   per revision: when it raises (the revision is listed in ``failed`` as
+   ``COMMIT_FAILED``), or when the receipt write after it fails, the next
+   reconcile calls it again, so make it idempotent on ``revision.resource``
+   (for example, key the load on the materialization id). Load the data, then return
+   :class:`CommitResult` (``accepted``, with ``consumer_commit_ref``),
+   ``"defer"`` (try again next run) or :class:`Reject` (``rejected``, with
+   your codes).
+
+Receipt ids are stable per (obligation, revision, materialization), and a
+materialization already receipted, accepted or rejected, is not receipted
+again.
+
+``mode="off"`` (the default) submits nothing; billing periods then show
+``receipt: {"status": "owed_by_buyer"}`` and ``responsible_party: "buyer"``,
+and the client logs a warning once. Adjustments are offered to
+``adjustments(adjustment)``; the default defers, and billing never
+auto-accepts. Accepting an adjustment needs its raw evidence, so it stays on
+:meth:`ReportingClient.submit_receipts`.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import inspect as _inspect
+import logging
+from collections.abc import Awaitable, Callable, Sequence
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any, Literal
+from uuid import uuid4
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class CommitResult:
+    """Your commit loaded the data: submit an ``accepted`` receipt with this reference."""
+
+    commit_ref: str
+
+
+@dataclass(frozen=True)
+class Reject:
+    """Your commit refused the data: submit a ``rejected`` receipt with these codes."""
+
+    codes: tuple[str, ...]
+
+    def __init__(self, codes: Sequence[str]) -> None:
+        if not codes:
+            raise ValueError("Reject needs at least one rejection code")
+        object.__setattr__(self, "codes", tuple(codes))
+
+
+CommitOutcome = CommitResult | Reject | Literal["defer"]
+
+
+@dataclass(frozen=True)
+class VerifiedRevision:
+    """What the SDK verified, handed to your ``commit`` hook."""
+
+    account_id: str
+    obligation: Any
+    revision: Any
+    materialization: Any
+    observation: Any
+
+    @property
+    def resource(self) -> Any:
+        """The materialized resource (``materialization.resource``)."""
+        return self.materialization.resource
+
+
+@dataclass(frozen=True)
+class ReceiptPolicy:
+    """How ``reconcile`` handles ``consumer_receipt`` revisions.
+
+    ``mode="verified"`` needs ``commit`` and one of ``resource_reader`` or
+    ``inspect``. ``adjustments`` sees each pending adjustment of a delivered
+    revision; return ``"defer"``.
+    """
+
+    mode: Literal["off", "verified"] = "off"
+    commit: Callable[[VerifiedRevision], Awaitable[CommitOutcome] | CommitOutcome] | None = None
+    adjustments: Callable[[Any], Awaitable[str] | str] | None = None
+    resource_reader: Any | None = None
+    inspect: Callable[[Any], Awaitable[Any]] | None = None
+
+    def __post_init__(self) -> None:
+        if self.mode not in ("off", "verified"):
+            raise ValueError("ReceiptPolicy mode is 'off' or 'verified'")
+        if self.mode == "verified":
+            if self.commit is None:
+                raise ValueError(
+                    "ReceiptPolicy(mode='verified') needs commit: the hook that loads the data "
+                    "and returns CommitResult, 'defer' or Reject."
+                )
+            if (self.resource_reader is None) == (self.inspect is None):
+                raise ValueError(
+                    "ReceiptPolicy(mode='verified') needs exactly one of resource_reader (the "
+                    "built-in manifest inspector) or inspect (your verifier)."
+                )
+
+
+@dataclass
+class ReceiptReport:
+    """What the receipts policy did in one ``reconcile``."""
+
+    submitted: list[Any] = field(default_factory=list)
+    rejected: list[Any] = field(default_factory=list)
+    deferred: list[str] = field(default_factory=list)
+    #: ``(reporting_obligation_id, code)`` for revisions the SDK could not verify:
+    #: the built-in inspector refused the materialization (a digest, row-count or
+    #: object mismatch) or a read failed. Nothing is receipted for them.
+    failed: list[tuple[str, str]] = field(default_factory=list)
+    adjustments_deferred: list[str] = field(default_factory=list)
+    owed_by_buyer: int = 0
+
+
+def _receipt_id(obligation: Any, revision: Any, materialization: Any) -> str:
+    """A stable receipt id per (obligation, revision, materialization), so a resend replays."""
+    digest = hashlib.sha256(
+        "\x1f".join(
+            (
+                obligation.reporting_obligation_id,
+                revision.reporting_revision_id,
+                materialization.reporting_materialization_id,
+            )
+        ).encode()
+    ).hexdigest()
+    return f"reporting-receipt:{digest[:40]}"
+
+
+async def _resolve(value: Any) -> Any:
+    return await value if _inspect.isawaitable(value) else value
+
+
+async def apply_receipt_policy(
+    client: Any,
+    account_id: str,
+    ledger: Any,
+    policy: ReceiptPolicy,
+    report: ReceiptReport,
+    *,
+    now: datetime,
+) -> bool:
+    """Verify, commit and receipt one account's unreceipted revisions; ``True`` when it wrote."""
+    from adcp.reporting._reconcile import (
+        ReportingInspectionContext,
+        ReportingObservation,
+        _enum,
+        _receipt_matches,
+        _validate_read_ledger,
+        build_reporting_receipt,
+    )
+    from adcp.types import SyncReportingReceiptsRequest
+
+    if policy.mode != "verified":
+        return False
+    inspect = policy.inspect
+    if inspect is None:
+        from adcp.reporting_inspection import ManifestReportingInspector
+
+        assert policy.resource_reader is not None
+        inspect = ManifestReportingInspector(policy.resource_reader)
+    commit = policy.commit
+    assert commit is not None
+    _leaves, _complete, index = _validate_read_ledger(ledger)
+    outgoing: list[Any] = []
+    for obligation in ledger.obligations:
+        if _enum(obligation.reconciliation_mode) != "consumer_receipt":
+            continue
+        revision, materialization, reasons = index.selections[obligation.reporting_obligation_id]
+        if revision is None or materialization is None or reasons:
+            continue
+        if any(_receipt_matches(item, revision, materialization) for item in ledger.receipts):
+            continue
+        if any(
+            item.reporting_obligation_id == obligation.reporting_obligation_id
+            and item.reporting_revision_id == revision.reporting_revision_id
+            and item.reporting_materialization_id == materialization.reporting_materialization_id
+            for item in ledger.receipts
+        ):
+            # Already receipted, rejected included: a refusal is stated once per
+            # materialization, never re-sent on every run.
+            continue
+        for adjustment in index.adjustments_by_revision.get(revision.reporting_revision_id, []):
+            decision = "defer"
+            if policy.adjustments is not None:
+                decision = str(await _resolve(policy.adjustments(adjustment)))
+            if decision != "defer":
+                logger.warning(
+                    "adjustment %s: only 'defer' is applied by the receipts policy; accept or "
+                    "reject it with client.reporting.submit_receipts",
+                    adjustment.reporting_adjustment_id,
+                )
+            report.adjustments_deferred.append(adjustment.reporting_adjustment_id)
+        context = ReportingInspectionContext(obligation, revision, materialization)
+        try:
+            observation = await inspect(context)
+        except Exception as error:  # noqa: BLE001 - one revision never fails the account
+            raw = getattr(error, "code", None)
+            code = "INSPECTION_FAILED" if raw is None else _enum(raw)
+            report.failed.append((obligation.reporting_obligation_id, code))
+            logger.warning(
+                "receipts: %s could not be verified (%s); nothing was receipted for it",
+                obligation.reporting_obligation_id,
+                code,
+            )
+            continue
+        receipt = build_reporting_receipt(
+            context,
+            observation,
+            reporting_receipt_id=_receipt_id(obligation, revision, materialization),
+            observed_at=now,
+        )
+        if _enum(receipt.status) == "rejected":
+            outgoing.append(receipt)
+            report.rejected.append(receipt)
+            continue
+        try:
+            decision_value = await _resolve(
+                commit(
+                    VerifiedRevision(account_id, obligation, revision, materialization, observation)
+                )
+            )
+        except Exception as error:  # noqa: BLE001 - one commit never drops the others
+            report.failed.append((obligation.reporting_obligation_id, "COMMIT_FAILED"))
+            logger.warning(
+                "receipts: commit failed for %s (%s); it is retried on the next reconcile",
+                obligation.reporting_obligation_id,
+                type(error).__name__,
+            )
+            continue
+        if isinstance(decision_value, CommitResult):
+            committed = ReportingObservation(
+                row_count=observation.row_count,
+                control_totals=observation.control_totals,
+                canonical_content_digest=observation.canonical_content_digest,
+                manifest_sha256=observation.manifest_sha256,
+                native_version_ref=observation.native_version_ref,
+                consumer_commit_ref=decision_value.commit_ref,
+            )
+            receipt = build_reporting_receipt(
+                context,
+                committed,
+                reporting_receipt_id=receipt.reporting_receipt_id,
+                observed_at=now,
+            )
+            outgoing.append(receipt)
+            report.submitted.append(receipt)
+        elif isinstance(decision_value, Reject):
+            payload = receipt.model_dump(mode="json", exclude_none=True)
+            payload.update(status="rejected", rejection_codes=list(decision_value.codes))
+            rejected = type(receipt).model_validate(payload)
+            outgoing.append(rejected)
+            report.rejected.append(rejected)
+        elif decision_value == "defer":
+            report.deferred.append(obligation.reporting_obligation_id)
+        else:
+            raise TypeError("commit returns CommitResult, Reject or 'defer'")
+    if not outgoing:
+        return False
+    write = await client.sync_reporting_receipts(
+        SyncReportingReceiptsRequest.model_validate(
+            {
+                "account": {"account_id": account_id},
+                "idempotency_key": f"rrcpt-{uuid4()}",
+                "receipts": [item.model_dump(mode="json", exclude_none=True) for item in outgoing],
+            }
+        )
+    )
+    data = getattr(write, "data", None)
+    if not getattr(write, "success", False) or data is None:
+        from adcp.reporting._reconcile import ReportingReconciliationError
+
+        raise ReportingReconciliationError(
+            "RECEIPT_WRITE_FAILED", "the seller did not record the reporting receipts; retry"
+        )
+    return True
+
+
+__all__ = [
+    "CommitOutcome",
+    "CommitResult",
+    "ReceiptPolicy",
+    "ReceiptReport",
+    "Reject",
+    "VerifiedRevision",
+    "apply_receipt_policy",
+]
diff --git a/src/adcp/reporting/subscriptions/reconcile.py b/src/adcp/reporting/subscriptions/reconcile.py
index 8c7d2b82e..b97c923a5 100644
--- a/src/adcp/reporting/subscriptions/reconcile.py
+++ b/src/adcp/reporting/subscriptions/reconcile.py
@@ -32,6 +32,7 @@
 
 from __future__ import annotations
 
+import logging
 from collections.abc import Iterable, Mapping, Sequence
 from dataclasses import dataclass, field
 from datetime import date, datetime, timedelta, timezone
@@ -51,8 +52,15 @@
     _Candidate,
     sort_traces,
 )
+from adcp.reporting.subscriptions.receipts import (
+    ReceiptPolicy,
+    ReceiptReport,
+    apply_receipt_policy,
+)
 from adcp.reporting.subscriptions.resolution import SubscriptionOffering
 
+logger = logging.getLogger(__name__)
+
 PeriodStatus = Literal["delivered", "pending", "late", "missing_obligation", "withheld", "not_owed"]
 ResponsibleParty = Literal["seller", "buyer", "none"]
 
@@ -169,6 +177,8 @@ class ReconcileReport:
     periods: tuple[PeriodLine, ...]
     issues: tuple[ReconcileIssue, ...] = ()
     accounts: tuple[str, ...] = ()
+    #: What the receipts policy did; ``None`` with ``mode="off"``.
+    receipts: ReceiptReport | None = None
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -250,6 +260,7 @@ class BuyerReportingState:
 
     def __init__(self) -> None:
         self.accounts: dict[str, _Account] = {}
+        self.warned_receipts = False
 
     def account(self, account_id: str) -> _Account:
         record = self.accounts.get(account_id)
@@ -599,7 +610,7 @@ class _Obligation:
 
 async def _obligations(
     client: Any, account_id: str, now: datetime
-) -> tuple[dict[tuple[Any, ...], _Obligation], bool]:
+) -> tuple[dict[tuple[Any, ...], _Obligation], bool, Any]:
     """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
@@ -637,7 +648,7 @@ async def _obligations(
             finality,
             bool(outcome and "MISSING_MATCHING_CONSUMER_RECEIPT" in outcome.reasons),
         )
-    return index, ledger._read_fingerprint is not None
+    return index, ledger._read_fingerprint is not None, ledger
 
 
 def _line(
@@ -751,11 +762,13 @@ async def reconcile_subscriptions(
     subscription_id: str | None = None,
     grace: timedelta = DEFAULT_GRACE,
     include_not_owed: bool = False,
+    receipts: ReceiptPolicy | None = None,
     now: datetime | None = None,
 ) -> ReconcileReport:
     """What every covered account is owed, and its status. See :meth:`ReportingClient.reconcile`."""
     report, _context = await _reconcile(
         client,
+        receipts=receipts,
         state=state,
         since=since,
         accounts=accounts,
@@ -777,11 +790,14 @@ async def _reconcile(
     grace: timedelta,
     include_not_owed: bool,
     now: datetime | None,
+    receipts: ReceiptPolicy | None = None,
 ) -> tuple[ReconcileReport, _Context]:
     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)
+    policy = receipts
+    receipt_report = ReceiptReport()
     context = _Context(client, state or BuyerReportingState())
     records = await context.load(accounts=None if accounts == "covered" else list(accounts))
     lines: list[PeriodLine] = []
@@ -791,7 +807,14 @@ async def _reconcile(
         obligations: dict[tuple[Any, ...], _Obligation] = {}
         if readable:
             try:
-                obligations, walked = await _obligations(client, record.account_id, instant)
+                obligations, walked, ledger = await _obligations(client, record.account_id, instant)
+                if policy is not None and policy.mode == "verified" and walked:
+                    if await apply_receipt_policy(
+                        client, record.account_id, ledger, policy, receipt_report, now=instant
+                    ):
+                        obligations, walked, ledger = await _obligations(
+                            client, record.account_id, instant
+                        )
                 complete = complete and walked
             except Exception as error:  # noqa: BLE001
                 complete = False
@@ -859,6 +882,18 @@ async def _reconcile(
         counts[line.status] += 1
         if line.status != "not_owed":
             counts["owed"] += 1
+    receipt_report.owed_by_buyer = sum(
+        1 for line in lines if line.receipt and line.receipt.get("status") == "owed_by_buyer"
+    )
+    if receipt_report.owed_by_buyer and (policy is None or policy.mode == "off"):
+        if not context.state.warned_receipts:
+            context.state.warned_receipts = True
+            logger.warning(
+                "%d billing period(s) owe a consumer receipt and the receipts policy is off. "
+                "Pass receipts=ReceiptPolicy(mode='verified', commit=...) or submit them with "
+                "client.reporting.submit_receipts.",
+                receipt_report.owed_by_buyer,
+            )
     report = ReconcileReport(
         as_of=iso_z(instant),
         definitive=complete and not context.issues,
@@ -866,6 +901,7 @@ async def _reconcile(
         periods=tuple(lines),
         issues=tuple(context.issues),
         accounts=tuple(record.account_id for record in records),
+        receipts=receipt_report if policy is not None and policy.mode != "off" else None,
     )
     return report, context
 
diff --git a/tests/test_reporting_subscription_receipts.py b/tests/test_reporting_subscription_receipts.py
new file mode 100644
index 000000000..5ae22b4a3
--- /dev/null
+++ b/tests/test_reporting_subscription_receipts.py
@@ -0,0 +1,203 @@
+"""The buyer's receipts policy: never on read, only after the SDK verified and you committed."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+from pydantic import TypeAdapter
+
+from adcp.reporting import ReportingObservation
+from adcp.reporting._reconcile import load_reporting_ledger
+from adcp.reporting.subscriptions import (
+    CommitResult,
+    ReceiptPolicy,
+    ReceiptReport,
+    Reject,
+    VerifiedRevision,
+)
+from adcp.reporting.subscriptions.facade import ReportingClient
+from adcp.reporting.subscriptions.receipts import apply_receipt_policy
+from adcp.types import GetReportingStatusRequest
+from adcp.types.domains.core.reporting_canonical_content_digest import (
+    ReportingCanonicalContentDigest,
+)
+from adcp.types.domains.core.reporting_control_total import ReportingControlTotal
+from tests.test_reporting_reconciliation import DIGEST, TOTALS, _Client
+
+pytestmark = pytest.mark.asyncio
+NOW = datetime(2026, 9, 3, tzinfo=timezone.utc)
+
+
+def _observation(rows: int = 7) -> ReportingObservation:
+    return ReportingObservation(
+        row_count=rows,
+        control_totals=[TypeAdapter(ReportingControlTotal).validate_python(i) for i in TOTALS],
+        canonical_content_digest=ReportingCanonicalContentDigest.model_validate(DIGEST),
+    )
+
+
+async def _ledger(client: _Client):
+    request = GetReportingStatusRequest.model_validate(
+        {"account": {"account_id": "account-1"}, "view": "periods"}
+    )
+    return await load_reporting_ledger(client, request)
+
+
+def test_the_policy_is_off_by_default_and_verified_needs_commit_and_a_verifier() -> None:
+    assert ReceiptPolicy().mode == "off"
+    with pytest.raises(ValueError, match="needs commit"):
+        ReceiptPolicy(mode="verified", inspect=lambda _c: None)
+    with pytest.raises(ValueError, match="exactly one"):
+        ReceiptPolicy(mode="verified", commit=lambda _r: "defer")
+    with pytest.raises(ValueError):
+        Reject([])
+
+
+async def test_a_commit_ref_submits_an_accepted_receipt_after_verification() -> None:
+    client = _Client()
+    seen: list[VerifiedRevision] = []
+
+    async def inspect(_context):
+        return _observation()
+
+    async def commit(revision: VerifiedRevision):
+        seen.append(revision)
+        return CommitResult("warehouse-job-42")
+
+    report = ReceiptReport()
+    policy = ReceiptPolicy(mode="verified", commit=commit, inspect=inspect)
+    wrote = await apply_receipt_policy(
+        client, "account-1", await _ledger(client), policy, report, now=NOW
+    )
+    assert wrote
+    assert [r.consumer_commit_ref for r in report.submitted] == ["warehouse-job-42"]
+    assert client.recorded_receipt["status"] == "accepted"
+    assert seen[0].observation.row_count == 7 and seen[0].resource is not None
+    # Receipted: a second run submits nothing.
+    again = ReceiptReport()
+    assert not await apply_receipt_policy(
+        client, "account-1", await _ledger(client), policy, again, now=NOW
+    )
+
+
+async def test_defer_submits_nothing_and_reject_sends_your_codes() -> None:
+    async def inspect(_context):
+        return _observation()
+
+    deferred = ReceiptReport()
+    client = _Client()
+    policy = ReceiptPolicy(mode="verified", commit=lambda _r: "defer", inspect=inspect)
+    assert not await apply_receipt_policy(
+        client, "account-1", await _ledger(client), policy, deferred, now=NOW
+    )
+    assert deferred.deferred == ["obligation-billing"] and client.recorded_receipt is None
+    rejected = ReceiptReport()
+    policy = ReceiptPolicy(
+        mode="verified", commit=lambda _r: Reject(["WAREHOUSE_LOAD_FAILED"]), inspect=inspect
+    )
+    assert await apply_receipt_policy(
+        client, "account-1", await _ledger(client), policy, rejected, now=NOW
+    )
+    assert client.recorded_receipt["status"] == "rejected"
+    assert client.recorded_receipt["rejection_codes"] == ["WAREHOUSE_LOAD_FAILED"]
+
+
+async def test_a_mismatch_the_sdk_found_is_rejected_without_calling_commit() -> None:
+    calls: list[object] = []
+
+    async def inspect(_context):
+        return _observation(rows=6)
+
+    client = _Client()
+    report = ReceiptReport()
+    policy = ReceiptPolicy(mode="verified", commit=calls.append, inspect=inspect)
+    assert await apply_receipt_policy(
+        client, "account-1", await _ledger(client), policy, report, now=NOW
+    )
+    assert calls == []
+    assert [r.rejection_codes for r in report.rejected] == [["ROW_COUNT_MISMATCH"]]
+
+
+async def test_off_submits_nothing() -> None:
+    client = _Client()
+    report = ReceiptReport()
+    assert not await apply_receipt_policy(
+        client, "account-1", await _ledger(client), ReceiptPolicy(), report, now=NOW
+    )
+    assert client.recorded_receipt is None
+
+
+async def test_submit_receipts_is_the_manual_path() -> None:
+    client = _Client()
+    reporting = ReportingClient(client)
+    from adcp.reporting import build_reporting_receipt
+    from adcp.reporting._reconcile import ReportingInspectionContext, _validate_read_ledger
+
+    ledger = await _ledger(client)
+    _leaves, _complete, index = _validate_read_ledger(ledger)
+    obligation = ledger.obligations[0]
+    revision, materialization, _ = index.selections[obligation.reporting_obligation_id]
+    receipt = build_reporting_receipt(
+        ReportingInspectionContext(obligation, revision, materialization), _observation()
+    )
+    results = await reporting.submit_receipts("account-1", receipts=[receipt])
+    assert [str(getattr(r.result, "value", r.result)) for r in results] == ["recorded"]
+    from adcp.reporting.subscriptions import ReportingSubscriptionError
+
+    with pytest.raises(ReportingSubscriptionError, match="Nothing to submit"):
+        await reporting.submit_receipts("account-1")
+
+
+async def test_a_rejection_is_sent_once_and_an_unverifiable_revision_is_not_receipted() -> None:
+    async def inspect(_context):
+        return _observation()
+
+    client = _Client()
+    policy = ReceiptPolicy(
+        mode="verified", commit=lambda _r: Reject(["LOAD_FAILED"]), inspect=inspect
+    )
+    first = ReceiptReport()
+    assert await apply_receipt_policy(
+        client, "account-1", await _ledger(client), policy, first, now=NOW
+    )
+    again = ReceiptReport()
+    assert not await apply_receipt_policy(
+        client, "account-1", await _ledger(client), policy, again, now=NOW
+    )
+    assert again.rejected == []
+
+    from adcp.reporting_inspection import ReportingInspectionError
+
+    async def refusing(_context):
+        from adcp.reporting_inspection import ReportingInspectionCode
+
+        raise ReportingInspectionError(
+            ReportingInspectionCode("MANIFEST_DIGEST_MISMATCH"), "the manifest does not match"
+        )
+
+    fresh = _Client()
+    failed = ReceiptReport()
+    policy = ReceiptPolicy(mode="verified", commit=lambda _r: "defer", inspect=refusing)
+    assert not await apply_receipt_policy(
+        fresh, "account-1", await _ledger(fresh), policy, failed, now=NOW
+    )
+    assert failed.failed == [("obligation-billing", "MANIFEST_DIGEST_MISMATCH")]
+    assert fresh.recorded_receipt is None
+
+
+async def test_a_commit_that_raises_is_retried_later_and_never_drops_other_receipts() -> None:
+    async def inspect(_context):
+        return _observation()
+
+    def broken(_revision):
+        raise TimeoutError("warehouse unavailable")
+
+    client = _Client()
+    report = ReceiptReport()
+    policy = ReceiptPolicy(mode="verified", commit=broken, inspect=inspect)
+    assert not await apply_receipt_policy(
+        client, "account-1", await _ledger(client), policy, report, now=NOW
+    )
+    assert report.failed == [("obligation-billing", "COMMIT_FAILED")]
+    assert client.recorded_receipt is None