fix: reporting re-grant chain, adjustment offers and rejected receipts match TS

docs/reporting-subscriptions-buyer.md

@@ -66,7 +66,7 @@ Each owed period that has ended becomes a `PeriodLine`, with wire-style keys and
 
 | `status` | Meaning | `responsible_party` |
 |---|---|---|
-| `delivered` | a current revision meets the selector's `required_finality` | `none`, or `buyer` when a billing receipt is owed (`receipt.status: owed_by_buyer`) |
+| `delivered` | a current revision meets the selector's `required_finality` | `none`; `buyer` when a billing receipt is owed (`receipt.status: owed_by_buyer`); `seller` once you rejected the revision (`receipt.status: rejected_by_buyer`: the seller corrects and re-delivers) |
 | `pending` | not yet due: within `grace` of the period end, or of `expected_at` | `none` |
 | `late` | an obligation exists, but its revision is more than `grace` past `expected_at` | `seller` |
 | `missing_obligation` | no obligation `grace` after the period ended | `seller` |
@@ -152,6 +152,7 @@ With `mode="verified"`, the SDK handles each `consumer_receipt` obligation whose
 
 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`.
+- **Rejected.** A period whose current revision you rejected reads `receipt: {"status": "rejected_by_buyer"}` and `responsible_party: "seller"`: the rejection is final for that materialization, and the seller corrects and re-delivers.
+- **Adjustments.** Each adjustment without an adjustment receipt is offered to `adjustments(adjustment)`, also on a revision already receipted, 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/events.py

@@ -519,6 +519,8 @@ async def poll(
         previous, revision = (None, None) if loaded is None else loaded
         if previous is not None and previous.get("snapshot_version") != SNAPSHOT_VERSION:
             previous = None
+        if previous is not None:
+            _remember_authorizations(previous, current)
         if not report.definitive and previous is not None:
             # Never treat an unread account as a change: keep the previous view of
             # what this run could not read.
@@ -598,6 +600,22 @@ async def handle_webhook(self, verified: Any) -> EventBatch | None:
         return await self.poll()
 
 
+def _remember_authorizations(previous: Mapping[str, Any], current: dict[str, Any]) -> None:
+    """Keep the last authorization of an account that is unauthorized now.
+
+    An unauthorized account is missing from ``list_accounts``, so the
+    snapshot taken during the loss would have no authorization for it, and
+    a re-grant would always read ``chain: "new"``, even one within
+    retention that keeps ``authorized_at``. The TS SDK does the same.
+    """
+    known = previous.get("authorizations") or {}
+    authorizations = dict(current.get("authorizations") or {})
+    for account_id in current.get("unauthorized") or {}:
+        if account_id not in authorizations and account_id in known:
+            authorizations[account_id] = known[account_id]
+    current["authorizations"] = authorizations
+
+
 def _merge_partial(
     previous: Mapping[str, Any], current: dict[str, Any], report: ReconcileReport
 ) -> dict[str, Any]:

src/adcp/reporting/subscriptions/receipts.py

@@ -30,9 +30,13 @@
 
 ``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
+and the client logs a warning once. A period whose current revision the buyer
+rejected reads ``receipt: {"status": "rejected_by_buyer"}`` and
+``responsible_party: "seller"``, which corrects and re-delivers it. 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
+auto-accepts. Each adjustment without an adjustment receipt is offered, also
+on a revision already receipted (an adjustment corrects a receipted revision).
+Accepting an adjustment needs its raw evidence, so it stays on
 :meth:`ReportingClient.submit_receipts`.
 """
 
@@ -183,25 +187,20 @@ async def apply_receipt_policy(
     commit = policy.commit
     assert commit is not None
     _leaves, _complete, index = _validate_read_ledger(ledger)
+    answered = {item.reporting_adjustment_id for item in ledger.adjustment_receipts}
     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
+        # An adjustment corrects a revision after the buyer receipted it, so the
+        # pending ones (no adjustment receipt yet) are offered whether or not the
+        # revision itself still needs a receipt.
         for adjustment in index.adjustments_by_revision.get(revision.reporting_revision_id, []):
+            if adjustment.reporting_adjustment_id in answered:
+                continue
             decision = "defer"
             if policy.adjustments is not None:
                 decision = str(await _resolve(policy.adjustments(adjustment)))
@@ -212,6 +211,17 @@ async def apply_receipt_policy(
                     adjustment.reporting_adjustment_id,
                 )
             report.adjustments_deferred.append(adjustment.reporting_adjustment_id)
+        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
         context = ReportingInspectionContext(obligation, revision, materialization)
         try:
             observation = await inspect(context)

src/adcp/reporting/subscriptions/reconcile.py

@@ -605,7 +605,10 @@ class _Obligation:
     obligation_id: str
     revision_id: str | None
     finality: str | None
-    receipt_owed: bool
+    #: ``owed_by_buyer`` until the buyer states anything about the current
+    #: revision, ``rejected_by_buyer`` once it rejected it, ``None`` when accepted
+    #: or not a billing (``consumer_receipt``) obligation.
+    receipt: str | None
 
 
 async def _obligations(
@@ -646,11 +649,36 @@ async def _obligations(
             obligation.reporting_obligation_id,
             None if revision is None else revision.reporting_revision_id,
             finality,
-            bool(outcome and "MISSING_MATCHING_CONSUMER_RECEIPT" in outcome.reasons),
+            _receipt_state(
+                obligation.reporting_obligation_id,
+                None if revision is None else revision.reporting_revision_id,
+                bool(outcome and "MISSING_MATCHING_CONSUMER_RECEIPT" in outcome.reasons),
+                ledger.receipts,
+            ),
         )
     return index, ledger._read_fingerprint is not None, ledger
 
 
+def _receipt_state(
+    obligation_id: str, revision_id: str | None, missing: bool, receipts: Iterable[Any]
+) -> str | None:
+    """Where a delivered billing revision's receipt stands, as in the TS SDK.
+
+    Only an accepted receipt clears ``MISSING_MATCHING_CONSUMER_RECEIPT``. A
+    rejected one is final (the receipts policy never resends it), so the
+    period is the seller's to correct and re-deliver, not the buyer's.
+    """
+    if not missing:
+        return None
+    rejected = any(
+        item.reporting_obligation_id == obligation_id
+        and item.reporting_revision_id == revision_id
+        and str(getattr(item.status, "value", item.status)) == "rejected"
+        for item in receipts
+    )
+    return "rejected_by_buyer" if rejected else "owed_by_buyer"
+
+
 def _line(
     period: OwedSubscriptionPeriod,
     trace_reason: str,
@@ -679,9 +707,13 @@ def _line(
         )
     ):
         status = "delivered"
-        if obligation.receipt_owed:
+        if obligation.receipt == "owed_by_buyer":
             responsible = "buyer"
             receipt = {"status": "owed_by_buyer"}
+        elif obligation.receipt == "rejected_by_buyer":
+            # The buyer rejected the current revision: the seller corrects it.
+            responsible = "seller"
+            receipt = {"status": "rejected_by_buyer"}
     elif period.withheld or trace_reason == "owed:withheld":
         # Withheld while authority is lost: never missing or late.
         status = "withheld"

tests/test_reporting_subscription_events.py

@@ -91,6 +91,35 @@ async def test_losing_and_regaining_authority_are_events() -> None:
     await world.service.drain_subscription_triggers()
     back = await reporting.poll(now=world.clock())
     assert "account.regranted" in _types(back)
+    regranted = next(e for e in back.events if e.type == "account.regranted")
+    # A re-grant within retention keeps authorized_at: the chain continues.
+    assert regranted.to_dict()["data"]["chain"] == "continued"
+    assert regranted.to_dict()["data"]["authorized_at"] == "2026-10-01T00:00:00Z"
+
+
+async def test_a_fresh_client_reads_the_chain_from_the_saved_snapshot() -> None:
+    from adcp.reporting.subscriptions import InMemoryReportingEventStore
+
+    world = _World()
+    await world.subscribe()
+    store = InMemoryReportingEventStore()
+
+    def fresh() -> ReportingClient:
+        return ReportingClient(InProcessBuyer(world.platform, _ctx), event_store=store)
+
+    world.clock.set(START + timedelta(minutes=30))
+    await (await fresh().poll(baseline=True, now=world.clock())).ack()
+    account = world.seller.accounts.pop(ACCOUNT)
+    await world.service.touch(account_id=ACCOUNT)
+    await world.service.drain_subscription_triggers()
+    await (await fresh().poll(now=world.clock())).ack()
+    world.clock.set(START + timedelta(hours=1))
+    world.seller.accounts[ACCOUNT] = account
+    await world.service.touch(account_id=ACCOUNT)
+    await world.service.drain_subscription_triggers()
+    back = await fresh().poll(now=world.clock())
+    regranted = next(e for e in back.events if e.type == "account.regranted")
+    assert regranted.to_dict()["data"]["chain"] == "continued"
 
 
 async def test_handle_webhook_needs_a_verified_delivery_and_polls_inline() -> None:

tests/test_reporting_subscription_receipts.py

@@ -201,3 +201,79 @@ def broken(_revision):
     )
     assert report.failed == [("obligation-billing", "COMMIT_FAILED")]
     assert client.recorded_receipt is None
+
+
+async def test_pending_adjustments_are_offered_on_an_already_receipted_revision(
+    monkeypatch,
+) -> None:
+    """An adjustment corrects a revision after it was receipted, so it is still offered."""
+    from types import SimpleNamespace
+
+    import adcp.reporting._reconcile as reconcile_module
+
+    async def inspect(_context):
+        return _observation()
+
+    client = _Client()
+    policy = ReceiptPolicy(
+        mode="verified", commit=lambda _r: CommitResult("job-1"), inspect=inspect
+    )
+    assert await apply_receipt_policy(
+        client, "account-1", await _ledger(client), policy, ReceiptReport(), now=NOW
+    )
+    ledger = await _ledger(client)
+    real = reconcile_module._validate_read_ledger
+
+    def with_adjustments(value):
+        leaves, complete, index = real(value)
+        obligation = value.obligations[0]
+        revision = index.selections[obligation.reporting_obligation_id][0]
+        index.adjustments_by_revision[revision.reporting_revision_id] = [
+            SimpleNamespace(reporting_adjustment_id="adjustment-pending"),
+            SimpleNamespace(reporting_adjustment_id="adjustment-answered"),
+        ]
+        value.adjustment_receipts.append(
+            SimpleNamespace(reporting_adjustment_id="adjustment-answered")
+        )
+        return leaves, complete, index
+
+    monkeypatch.setattr(reconcile_module, "_validate_read_ledger", with_adjustments)
+    seen: list[str] = []
+    offered = ReceiptPolicy(
+        mode="verified",
+        commit=lambda _r: CommitResult("job-2"),
+        inspect=inspect,
+        adjustments=lambda adjustment: seen.append(adjustment.reporting_adjustment_id) or "defer",
+    )
+    report = ReceiptReport()
+    assert not await apply_receipt_policy(client, "account-1", ledger, offered, report, now=NOW)
+    assert seen == report.adjustments_deferred == ["adjustment-pending"]
+    assert report.submitted == []
+
+
+async def test_a_rejected_revision_is_the_sellers_to_correct_not_owed_by_the_buyer() -> None:
+    from adcp.reporting.subscriptions.reconcile import _obligations
+
+    async def inspect(_context):
+        return _observation()
+
+    async def receipt_state(client: _Client) -> list[str | None]:
+        index, walked, _ledger_read = await _obligations(client, "account-1", NOW)
+        assert walked
+        return [item.receipt for item in index.values() if item.revision_id is not None]
+
+    client = _Client()
+    assert await receipt_state(client) == ["owed_by_buyer"]
+    policy = ReceiptPolicy(
+        mode="verified", commit=lambda _r: Reject(["LOAD_FAILED"]), inspect=inspect
+    )
+    await apply_receipt_policy(
+        client, "account-1", await _ledger(client), policy, ReceiptReport(), now=NOW
+    )
+    assert await receipt_state(client) == ["rejected_by_buyer"]
+    accepted = _Client()
+    policy = ReceiptPolicy(mode="verified", commit=lambda _r: CommitResult("job"), inspect=inspect)
+    await apply_receipt_policy(
+        accepted, "account-1", await _ledger(accepted), policy, ReceiptReport(), now=NOW
+    )
+    assert await receipt_state(accepted) == [None]