From c052344a97c97920d90c12c4fcf61f69f71a5a66 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 00:20:10 +0000 Subject: [PATCH] fix: reporting re-grant chain, adjustment offers and rejected receipts match TS Three parity notes from the TS ports of the buyer reporting slices: - events: an unauthorized account is missing from list_accounts, so the snapshot taken during the loss had no authorization for it and a re-grant always read chain new. poll now keeps the previous authorization of an account that is unauthorized now, so a re-grant within retention reads continued, also across fresh clients. - receipts: pending adjustments, those without an adjustment receipt, are offered before the already-receipted check. An adjustment corrects a revision after it was receipted, so the hook never saw one before. - reconcile: a delivered billing period whose current revision the buyer rejected reads receipt rejected_by_buyer with responsible_party seller, instead of owed_by_buyer forever. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reporting-subscriptions-buyer.md | 5 +- src/adcp/reporting/subscriptions/events.py | 18 +++++ src/adcp/reporting/subscriptions/receipts.py | 36 +++++---- src/adcp/reporting/subscriptions/reconcile.py | 38 +++++++++- tests/test_reporting_subscription_events.py | 29 +++++++ tests/test_reporting_subscription_receipts.py | 76 +++++++++++++++++++ 6 files changed, 184 insertions(+), 18 deletions(-) diff --git a/docs/reporting-subscriptions-buyer.md b/docs/reporting-subscriptions-buyer.md index 14f52baef..b4305c251 100644 --- a/docs/reporting-subscriptions-buyer.md +++ b/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. diff --git a/src/adcp/reporting/subscriptions/events.py b/src/adcp/reporting/subscriptions/events.py index 0dcd3aa3c..b5d4adac5 100644 --- a/src/adcp/reporting/subscriptions/events.py +++ b/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]: diff --git a/src/adcp/reporting/subscriptions/receipts.py b/src/adcp/reporting/subscriptions/receipts.py index bfb4c4d7c..082bb3071 100644 --- a/src/adcp/reporting/subscriptions/receipts.py +++ b/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,6 +187,7 @@ 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": @@ -190,18 +195,12 @@ async def apply_receipt_policy( 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) diff --git a/src/adcp/reporting/subscriptions/reconcile.py b/src/adcp/reporting/subscriptions/reconcile.py index b97c923a5..d0859c87d 100644 --- a/src/adcp/reporting/subscriptions/reconcile.py +++ b/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" diff --git a/tests/test_reporting_subscription_events.py b/tests/test_reporting_subscription_events.py index e6d117977..b73f04e8b 100644 --- a/tests/test_reporting_subscription_events.py +++ b/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: diff --git a/tests/test_reporting_subscription_receipts.py b/tests/test_reporting_subscription_receipts.py index 5ae22b4a3..7c5a487df 100644 --- a/tests/test_reporting_subscription_receipts.py +++ b/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]