feat(reporting): derive buyer reporting events with poll, events and handle_webhook

README.md

@@ -373,7 +373,8 @@ outcome = await client.accounts.ensure(key, billing="operator")  # AccountReady
 - **Storage.** One object; process-local storage is refused in production. Apply schema changes with `python -m adcp migrate`.
 - **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.
-- **Coming in later betas:** buyer events and the test kit.
+- **Events.** `client.reporting.poll()` / `events()` derive events from saved state with deterministic ids; `handle_webhook()` treats a verified webhook as a doorbell.
+- **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

@@ -93,7 +93,41 @@ traces = await client.reporting.explain(account_id="acc_a", subscription_id="ana
 
 ## Events
 
-`client.reporting.poll()`, `events()` and `handle_webhook()` arrive in a later beta. Until then, pass each `principal.changed` payload to `tracker.handle_notification(...)`.
+```python
+# A long-running worker: each batch is acked once the loop body ran for all of it.
+async for event in client.reporting.events(poll_every=timedelta(minutes=5)):
+    match event.type:
+        case "revision.available": await ingest(event.reporting_revision_id)
+        case "period.late": await alert_seller(event)
+        case "account.unauthorized": await pause_dashboards(event.account_id)
+
+# Serverless: one batch per invocation.
+batch = await client.reporting.poll()
+await handle(batch.events)
+await batch.ack()
+
+# Webhook endpoint: a doorbell, never the event content.
+outcome = await receiver.receive(...)               # verifies the signature
+await client.reporting.handle_webhook(VerifiedWebhook.from_outcome(outcome))
+```
+
+- **Derived, not pushed.** Each poll runs `reconcile` and reads the subscription readback and coverage, then compares the result with the snapshot saved at the last `ack`. A crash before `ack` redelivers the same events with the same ids, so deliver at least once and deduplicate on `event.id`.
+- **Deterministic ids.** `event.id` is `sha256(agent, type, natural key, version)`.
+- **Durable state.** Pass `ReportingClient(client, event_store=...)` for a store that survives restarts. It implements `load(agent_id)` and a compare-and-set `save(...)`. The default `InMemoryReportingEventStore` is process-local. Two workers that ack the same snapshot get `ReportingEventConflictError` on the second ack.
+- **Cold start.** `poll(baseline=True)` on a consumer with no snapshot records the present without emitting it.
+- **Webhooks.** `handle_webhook` takes a `VerifiedWebhook` (or a verified, non-rejected `WebhookOutcome`) and refuses anything else. With an `events()` loop running it wakes the loop; otherwise it polls and returns the batch for you to handle and `ack`.
+- **Partial reads.** An account that could not be read keeps its previous state, so it never produces a spurious "closed" or "gone" event.
+
+| Event | When | Key fields |
+|---|---|---|
+| `revision.available` | a revision appears for an owed period | `reporting_revision_id`, `reporting_obligation_id`, `period`, `finality` |
+| `revision.adjusted` | a new revision replaces one you saw | `superseded_reporting_revision_id` |
+| `period.late`, `period.missing_obligation`, `period.withheld` | a period reaches that status | the period line |
+| `coverage.exception_opened`, `coverage.exception_closed` | an `excluded[]` entry opens or closes | `reason`, `from`, `until`, `media_buy_id` |
+| `account.covered` | an account enters a subscription | `subscription_id`, `account_id` |
+| `account.unauthorized`, `account.regranted` | authority is lost, or returns | `unauthorized_at`, `regranted_at`, `chain` |
+| `subscription.state_changed` | `active`, `inactive` or `action_required` | `state`, `previous_state`, `pending_action` |
+| `timezone.cutover` | a `timezone_cutovers[]` entry appears | `effective_at`, `timezone`, `previous_timezone` |
 
 ## Receipts for billing feeds
 

src/adcp/reporting/subscriptions/__init__.py

@@ -81,6 +81,14 @@
     decode_principal_state,
     encode_principal_state,
 )
+from adcp.reporting.subscriptions.events import (
+    EventBatch,
+    InMemoryReportingEventStore,
+    ReportingEvent,
+    ReportingEventConflictError,
+    ReportingEventStore,
+    VerifiedWebhook,
+)
 from adcp.reporting.subscriptions.explain import (
     EXPLAIN_TEMPLATES,
     TRACE_VERSION,
@@ -140,6 +148,12 @@
 from adcp.server.reporting_grants import REPORTING_WIND_DOWN_TASKS, wind_down_grant
 
 __all__ = [
+    "EventBatch",
+    "InMemoryReportingEventStore",
+    "ReportingEvent",
+    "ReportingEventConflictError",
+    "ReportingEventStore",
+    "VerifiedWebhook",
     "BuyerReportingState",
     "CoverageExplainer",
     "PeriodLine",