From f0e8c9d1e072e7ca440c221717913daffabd9fc4 Mon Sep 17 00:00:00 2001
From: Brian O'Kelley <bokelley@scope3.com>
Date: Sun, 1 Nov 2026 17:30:11 +0000
Subject: [PATCH 1/5] feat(reporting): derive buyer reporting events with poll,
 events and handle_webhook

client.reporting.poll runs reconcile, reads the subscription readback and
coverage, and compares the result with the snapshot saved at the last ack,
emitting one event per difference: revision.available and revision.adjusted,
period.late, period.missing_obligation and period.withheld, coverage
exception opened and closed, account covered, unauthorized and regranted,
subscription.state_changed and timezone.cutover. batch.ack saves the
snapshot under compare-and-set, so a crash before ack redelivers the same
events, and event ids are sha256 of agent, type, natural key and version.
An account that could not be read keeps its previous state.

client.reporting.events is the async-iterator form, acked per batch, and
handle_webhook takes only a verified delivery, waking a running loop or
polling inline. The event store is pluggable; the default is in memory.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 README.md                                     |   3 +-
 docs/reporting-subscriptions-buyer.md         |  36 +-
 src/adcp/reporting/subscriptions/__init__.py  |  14 +
 src/adcp/reporting/subscriptions/events.py    | 621 ++++++++++++++++++
 src/adcp/reporting/subscriptions/facade.py    |  59 ++
 src/adcp/reporting/subscriptions/reconcile.py |  27 +-
 tests/test_reporting_subscription_events.py   | 137 ++++
 7 files changed, 894 insertions(+), 3 deletions(-)
 create mode 100644 src/adcp/reporting/subscriptions/events.py
 create mode 100644 tests/test_reporting_subscription_events.py

diff --git a/README.md b/README.md
index df1e8abdc..1176949c1 100644
--- a/README.md
+++ b/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)
 
diff --git a/docs/reporting-subscriptions-buyer.md b/docs/reporting-subscriptions-buyer.md
index 886e1a260..588ae8543 100644
--- a/docs/reporting-subscriptions-buyer.md
+++ b/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
 
diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py
index 0dd9751db..7a405a0a8 100644
--- a/src/adcp/reporting/subscriptions/__init__.py
+++ b/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",
diff --git a/src/adcp/reporting/subscriptions/events.py b/src/adcp/reporting/subscriptions/events.py
new file mode 100644
index 000000000..14ddd88ec
--- /dev/null
+++ b/src/adcp/reporting/subscriptions/events.py
@@ -0,0 +1,621 @@
+"""Buyer events for reporting subscriptions: ``poll``, ``events``, ``handle_webhook`` (AdCP 3.3).
+
+Events are derived from durable state, never taken from a webhook body:
+
+1. :meth:`ReportingClient.poll` runs ``reconcile`` and reads the subscription
+   readback and each account's ``subscription_coverage[]``.
+2. It compares that snapshot with the one this consumer last acknowledged,
+   and emits one :class:`ReportingEvent` per difference.
+3. ``await batch.ack()`` saves the new snapshot. Until then nothing moves:
+   a crash before the ack replays the same events, with the same ids, on the
+   next poll. Delivery is at least once; deduplicate on ``event.id``.
+
+Event ids are ``sha256(agent, type, natural key, version)``, so a replay
+after a crash, or the same change seen by two workers, produces the same id.
+
+A seller's ``principal.changed`` webhook is a doorbell: pass the verified
+delivery to :meth:`ReportingClient.handle_webhook`, which wakes a running
+:meth:`ReportingClient.events` loop, or polls inline when none runs.
+
+Event types (every event has ``id``, ``type``, ``agent_id``, ``occurred_at``,
+``observed_at`` and a ``data`` mapping):
+
+* ``revision.available`` and ``revision.adjusted`` (a new revision replaced
+  one this consumer saw);
+* ``period.late``, ``period.missing_obligation``, ``period.withheld``;
+* ``coverage.exception_opened`` and ``coverage.exception_closed``, with
+  ``reason``, ``from`` and ``until``;
+* ``account.covered``, ``account.unauthorized``, ``account.regranted``;
+* ``subscription.state_changed`` (``active``, ``inactive``,
+  ``action_required``);
+* ``timezone.cutover``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+from collections.abc import AsyncIterator, Mapping
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any, Protocol
+
+from adcp.reporting.canonical_json import canonical_json_utf8_v1
+from adcp.reporting.subscriptions._clock import iso_z, utc_instant
+from adcp.reporting.subscriptions.reconcile import (
+    DEFAULT_GRACE,
+    BuyerReportingState,
+    ReconcileReport,
+    _reconcile,
+)
+
+#: Version of the saved snapshot document; bump on a breaking change.
+SNAPSHOT_VERSION = 1
+
+EVENT_TYPES = frozenset(
+    {
+        "revision.available",
+        "revision.adjusted",
+        "period.late",
+        "period.missing_obligation",
+        "period.withheld",
+        "coverage.exception_opened",
+        "coverage.exception_closed",
+        "account.covered",
+        "account.unauthorized",
+        "account.regranted",
+        "subscription.state_changed",
+        "timezone.cutover",
+    }
+)
+
+
+@dataclass(frozen=True)
+class ReportingEvent:
+    """One change this consumer has not acknowledged yet."""
+
+    id: str
+    type: str
+    agent_id: str
+    occurred_at: str
+    observed_at: str
+    data: Mapping[str, Any] = field(default_factory=dict)
+
+    def __getattr__(self, name: str) -> Any:
+        # ``event.account_id``, ``event.reporting_revision_id`` and friends.
+        data = object.__getattribute__(self, "data")
+        if name in data:
+            return data[name]
+        raise AttributeError(name)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "id": self.id,
+            "type": self.type,
+            "agent_id": self.agent_id,
+            "occurred_at": self.occurred_at,
+            "observed_at": self.observed_at,
+            "data": dict(self.data),
+        }
+
+
+class ReportingEventStore(Protocol):
+    """Where a consumer's acknowledged snapshot lives, per agent.
+
+    ``save`` must refuse (raise) when the stored revision is no longer
+    ``expected_revision``, so two workers never both acknowledge.
+    """
+
+    async def load(self, agent_id: str) -> tuple[Mapping[str, Any], int] | None: ...
+
+    async def save(
+        self, agent_id: str, snapshot: Mapping[str, Any], *, expected_revision: int | None
+    ) -> int: ...
+
+
+class ReportingEventConflictError(RuntimeError):
+    """Another worker acknowledged first; poll again."""
+
+    code = "REPORTING_EVENTS_CONFLICT"
+    recovery = "transient"
+
+
+class InMemoryReportingEventStore:
+    """Process-local snapshots: development and tests only."""
+
+    def __init__(self) -> None:
+        self._items: dict[str, tuple[dict[str, Any], int]] = {}
+
+    async def load(self, agent_id: str) -> tuple[Mapping[str, Any], int] | None:
+        item = self._items.get(agent_id)
+        return None if item is None else (json.loads(json.dumps(item[0])), item[1])
+
+    async def save(
+        self, agent_id: str, snapshot: Mapping[str, Any], *, expected_revision: int | None
+    ) -> int:
+        current = self._items.get(agent_id)
+        if (current[1] if current else None) != expected_revision:
+            raise ReportingEventConflictError(
+                "Another worker acknowledged these events first. Poll again."
+            )
+        revision = (current[1] if current else 0) + 1
+        self._items[agent_id] = (json.loads(json.dumps(snapshot)), revision)
+        return revision
+
+
+@dataclass
+class EventBatch:
+    """One poll: the events since the last acknowledgement, and its ``ack``."""
+
+    events: list[ReportingEvent]
+    report: ReconcileReport
+    _store: ReportingEventStore
+    _agent_id: str
+    _snapshot: Mapping[str, Any]
+    _revision: int | None
+    acked: bool = False
+
+    async def ack(self) -> None:
+        """Save this poll's snapshot, so its events are not delivered again."""
+        if self.acked:
+            return
+        await self._store.save(self._agent_id, self._snapshot, expected_revision=self._revision)
+        self.acked = True
+
+
+class VerifiedWebhook:
+    """A webhook delivery whose signature the SDK verified.
+
+    Build it with :meth:`from_outcome` from a :class:`~adcp.webhook_receiver.WebhookReceiver`
+    outcome; :meth:`ReportingClient.handle_webhook` refuses anything else at
+    runtime, so an unverified body never reaches the poller.
+    """
+
+    __slots__ = ("payload", "sender_identity")
+
+    def __init__(self, payload: Any, sender_identity: str, *, _token: object = None) -> None:
+        if _token is not _VERIFIED:
+            raise TypeError("use VerifiedWebhook.from_outcome(outcome)")
+        self.payload = payload
+        self.sender_identity = sender_identity
+
+    @classmethod
+    def from_outcome(cls, outcome: Any) -> VerifiedWebhook:
+        """From a ``WebhookReceiver.receive`` outcome that verified and was not rejected."""
+        if getattr(outcome, "rejected", True) or not getattr(outcome, "sender_identity", None):
+            raise ValueError("the webhook was not verified; it cannot wake the poller")
+        return cls(outcome.payload, outcome.sender_identity, _token=_VERIFIED)
+
+
+_VERIFIED = object()
+
+
+# -- snapshots -------------------------------------------------------------------------
+
+
+def _key(*parts: Any) -> str:
+    return json.dumps([str(part) if part is not None else None for part in parts])
+
+
+def _snapshot(report: ReconcileReport, context: Any) -> dict[str, Any]:
+    """The state events are derived from, as a JSON document."""
+    periods: dict[str, dict[str, Any]] = {}
+    for line in report.periods:
+        if line.status == "not_owed":
+            continue
+        periods[
+            _key(
+                line.account_id,
+                line.subscription_id,
+                line.subscription_version,
+                line.offering_id,
+                line.period["start"],
+                line.period["end"],
+            )
+        ] = {
+            "status": line.status,
+            "reporting_revision_id": line.reporting_revision_id,
+            "reporting_obligation_id": line.reporting_obligation_id,
+            "finality": line.finality,
+            "expected_at": line.expected_at,
+            "period": dict(line.period),
+            "account_id": line.account_id,
+            "subscription_id": line.subscription_id,
+            "subscription_version": line.subscription_version,
+            "offering_id": line.offering_id,
+            "media_buy_ids": list(line.media_buy_ids),
+        }
+    subscriptions: dict[str, dict[str, Any]] = {}
+    unauthorized: dict[str, str] = {}
+    for state in context.subscriptions:
+        value = state.model_dump(mode="json", exclude_none=True)
+        issues = value.get("issues") or []
+        subscriptions[str(state.subscription_id)] = {
+            "state": value.get("state"),
+            "subscription_version": value.get("subscription_version"),
+            "pending_action": issues[0] if issues else None,
+        }
+        for entry in value.get("accounts_unauthorized") or []:
+            unauthorized[entry["account_id"]] = entry["unauthorized_at"]
+    covered: list[str] = sorted(
+        {_key(line.subscription_id, line.account_id) for line in report.periods}
+    )
+    exceptions: dict[str, dict[str, Any]] = {}
+    cutovers: dict[str, dict[str, Any]] = {}
+    authorizations: dict[str, dict[str, Any]] = {}
+    for account_id in report.accounts:
+        record = context.state.accounts.get(account_id)
+        if record is None:
+            continue
+        if record.listed:
+            authorizations[account_id] = {
+                "authorized_at": None
+                if record.authorized_at is None
+                else iso_z(record.authorized_at),
+                "regranted_at": None if record.regranted_at is None else iso_z(record.regranted_at),
+            }
+        for subscription_id, entry in record.coverage.items():
+            covered.append(_key(subscription_id, account_id))
+            for item in entry.get("excluded") or ():
+                exceptions[
+                    _key(
+                        subscription_id,
+                        account_id,
+                        item.get("reason"),
+                        item.get("media_buy_id"),
+                        item.get("from"),
+                        ",".join(item.get("package_ids") or ()),
+                    )
+                ] = {
+                    "subscription_id": subscription_id,
+                    "account_id": account_id,
+                    "reason": item.get("reason"),
+                    "media_buy_id": item.get("media_buy_id"),
+                    "from": item.get("from"),
+                    "until": item.get("until"),
+                    "package_ids": list(item.get("package_ids") or ()),
+                }
+            for item in entry.get("timezone_cutovers") or ():
+                cutovers[
+                    _key(
+                        subscription_id,
+                        account_id,
+                        item.get("offering_id"),
+                        item.get("effective_at"),
+                    )
+                ] = {
+                    "subscription_id": subscription_id,
+                    "account_id": account_id,
+                    "offering_id": item.get("offering_id"),
+                    "effective_at": item.get("effective_at"),
+                    "timezone": item.get("timezone"),
+                    "previous_timezone": item.get("previous_timezone"),
+                }
+    return {
+        "snapshot_version": SNAPSHOT_VERSION,
+        "periods": periods,
+        "subscriptions": subscriptions,
+        "covered": sorted(set(covered)),
+        "unauthorized": unauthorized,
+        "authorizations": authorizations,
+        "exceptions": exceptions,
+        "cutovers": cutovers,
+    }
+
+
+def _event_id(agent_id: str, type_: str, natural_key: str, version: Any) -> str:
+    digest = hashlib.sha256(
+        canonical_json_utf8_v1([agent_id, type_, natural_key, version])
+    ).hexdigest()
+    return f"rse_{digest[:40]}"
+
+
+def derive_events(
+    previous: Mapping[str, Any] | None,
+    current: Mapping[str, Any],
+    *,
+    agent_id: str,
+    observed_at: datetime,
+    grace: timedelta = DEFAULT_GRACE,
+) -> list[ReportingEvent]:
+    """The events between two snapshots, in a deterministic order.
+
+    With no ``previous`` snapshot every present fact is new: each delivered
+    revision, open exception, covered account and so on is emitted once.
+    """
+    before: Mapping[str, Any] = previous or {}
+    observed = iso_z(observed_at)
+    events: list[ReportingEvent] = []
+
+    def emit(
+        type_: str, natural_key: str, version: Any, occurred: str | None, data: Mapping[str, Any]
+    ) -> None:
+        events.append(
+            ReportingEvent(
+                id=_event_id(agent_id, type_, natural_key, version),
+                type=type_,
+                agent_id=agent_id,
+                occurred_at=occurred or observed,
+                observed_at=observed,
+                data=dict(data),
+            )
+        )
+
+    old_subscriptions = before.get("subscriptions") or {}
+    for subscription_id, value in sorted((current.get("subscriptions") or {}).items()):
+        old = old_subscriptions.get(subscription_id)
+        if old is None or old.get("state") != value.get("state"):
+            emit(
+                "subscription.state_changed",
+                _key(subscription_id),
+                [value.get("subscription_version"), value.get("state")],
+                None,
+                {
+                    "subscription_id": subscription_id,
+                    "state": value.get("state"),
+                    "previous_state": None if old is None else old.get("state"),
+                    "subscription_version": value.get("subscription_version"),
+                    "pending_action": value.get("pending_action"),
+                },
+            )
+
+    old_covered = set(before.get("covered") or ())
+    for item in current.get("covered") or ():
+        if item not in old_covered:
+            subscription_id, account_id = json.loads(item)
+            emit(
+                "account.covered",
+                item,
+                1,
+                None,
+                {"subscription_id": subscription_id, "account_id": account_id},
+            )
+
+    old_unauthorized = before.get("unauthorized") or {}
+    new_unauthorized = current.get("unauthorized") or {}
+    for account_id, lost in sorted(new_unauthorized.items()):
+        if old_unauthorized.get(account_id) != lost:
+            emit(
+                "account.unauthorized",
+                _key(account_id),
+                lost,
+                lost,
+                {"account_id": account_id, "unauthorized_at": lost},
+            )
+    old_authorizations = before.get("authorizations") or {}
+    for account_id, lost in sorted(old_unauthorized.items()):
+        if account_id in new_unauthorized:
+            continue
+        granted = (current.get("authorizations") or {}).get(account_id)
+        if granted is None:
+            continue
+        regranted = granted.get("regranted_at") or granted.get("authorized_at")
+        emit(
+            "account.regranted",
+            _key(account_id),
+            [lost, regranted],
+            regranted,
+            {
+                "account_id": account_id,
+                "unauthorized_at": lost,
+                "regranted_at": regranted,
+                "authorized_at": granted.get("authorized_at"),
+                "chain": "continued"
+                if granted.get("authorized_at")
+                == (old_authorizations.get(account_id) or {}).get("authorized_at")
+                else "new",
+            },
+        )
+
+    old_exceptions = before.get("exceptions") or {}
+    new_exceptions = current.get("exceptions") or {}
+    for key, value in sorted(new_exceptions.items()):
+        old = old_exceptions.get(key)
+        if value.get("until") is None and (old is None or old.get("until") is not None):
+            emit("coverage.exception_opened", key, value.get("from"), value.get("from"), value)
+        if value.get("until") is not None and (old is not None and old.get("until") is None):
+            emit("coverage.exception_closed", key, value.get("until"), value.get("until"), value)
+    for key, old in sorted(old_exceptions.items()):
+        if key not in new_exceptions and old.get("until") is None:
+            emit("coverage.exception_closed", key, "gone", None, {**old, "until": None})
+
+    old_cutovers = before.get("cutovers") or {}
+    for key, value in sorted((current.get("cutovers") or {}).items()):
+        if key not in old_cutovers:
+            emit("timezone.cutover", key, 1, value.get("effective_at"), value)
+
+    old_periods = before.get("periods") or {}
+    for key, value in sorted((current.get("periods") or {}).items()):
+        old = old_periods.get(key) or {}
+        line = {k: value[k] for k in value if k not in {"status"}}
+        revision = value.get("reporting_revision_id")
+        old_revision = old.get("reporting_revision_id")
+        if revision is not None and revision != old_revision:
+            type_ = "revision.adjusted" if old_revision is not None else "revision.available"
+            emit(
+                type_,
+                key,
+                revision,
+                None,
+                {**line, "superseded_reporting_revision_id": old_revision},
+            )
+        status = value.get("status")
+        if status != old.get("status") and status in {"late", "missing_obligation", "withheld"}:
+            occurred = None
+            if status == "late" and value.get("expected_at"):
+                occurred = iso_z(_parse(value["expected_at"]) + grace)
+            elif status == "missing_obligation":
+                occurred = iso_z(_parse(value["period"]["end"]) + grace)
+            emit(f"period.{status}", key, status, occurred, line)
+    return events
+
+
+def _parse(value: str) -> datetime:
+    return utc_instant(datetime.fromisoformat(value.replace("Z", "+00:00")))
+
+
+# -- the consumer ----------------------------------------------------------------------
+
+
+class ReportingEvents:
+    """``poll``, ``events`` and ``handle_webhook`` for one client (see the module doc)."""
+
+    def __init__(
+        self,
+        client: Any,
+        state: BuyerReportingState,
+        *,
+        store: ReportingEventStore | None = None,
+        agent_id: str | None = None,
+    ) -> None:
+        self._client = client
+        self._state = state
+        self.store: ReportingEventStore = store or InMemoryReportingEventStore()
+        self._agent_id = agent_id
+        self._doorbell: asyncio.Event | None = None
+        self._listening = 0
+
+    @property
+    def agent_id(self) -> str:
+        if self._agent_id is not None:
+            return self._agent_id
+        config = getattr(self._client, "agent_config", None)
+        identifier = getattr(config, "id", None) or getattr(config, "agent_uri", None)
+        return str(identifier or "default")
+
+    async def poll(
+        self,
+        *,
+        grace: timedelta = DEFAULT_GRACE,
+        baseline: bool = False,
+        now: datetime | None = None,
+    ) -> EventBatch:
+        """One batch of events since the last acknowledgement; call ``batch.ack()`` after handling.
+
+        ``baseline=True`` on a consumer with no snapshot yet records the
+        current state without emitting it.
+        """
+        instant = utc_instant(now, field="now") if now is not None else datetime.now(timezone.utc)
+        report, context = await _reconcile(
+            self._client,
+            state=self._state,
+            since=None,
+            accounts="covered",
+            subscription_id=None,
+            grace=grace,
+            include_not_owed=False,
+            now=instant,
+        )
+        current = _snapshot(report, context)
+        loaded = await self.store.load(self.agent_id)
+        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 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.
+            current = _merge_partial(previous, current, report)
+        if loaded is None and baseline:
+            events: list[ReportingEvent] = []
+        else:
+            events = derive_events(
+                previous, current, agent_id=self.agent_id, observed_at=instant, grace=grace
+            )
+        return EventBatch(events, report, self.store, self.agent_id, current, revision)
+
+    async def events(
+        self,
+        *,
+        poll_every: timedelta = timedelta(minutes=5),
+        stop: asyncio.Event | None = None,
+        grace: timedelta = DEFAULT_GRACE,
+    ) -> AsyncIterator[ReportingEvent]:
+        """Every event, forever (or until ``stop`` is set).
+
+        Each batch is acknowledged once the loop body has run for all of its
+        events, so an exception in the body redelivers the batch.
+        """
+        if self._doorbell is None:
+            self._doorbell = asyncio.Event()
+        self._listening += 1
+        try:
+            while stop is None or not stop.is_set():
+                batch = await self.poll(grace=grace)
+                for event in batch.events:
+                    yield event
+                await batch.ack()
+                self._doorbell.clear()
+                waiters = [asyncio.ensure_future(self._doorbell.wait())]
+                if stop is not None:
+                    waiters.append(asyncio.ensure_future(stop.wait()))
+                done, pending = await asyncio.wait(
+                    waiters,
+                    timeout=poll_every.total_seconds(),
+                    return_when=asyncio.FIRST_COMPLETED,
+                )
+                for item in pending:
+                    item.cancel()
+        finally:
+            self._listening -= 1
+
+    async def handle_webhook(self, verified: Any) -> EventBatch | None:
+        """A verified seller webhook: wake the ``events()`` loop, or poll inline.
+
+        ``verified`` is a :class:`VerifiedWebhook` or a
+        :class:`~adcp.webhook_receiver.WebhookOutcome` that verified and was not
+        rejected; anything else is refused. The body is never trusted as event
+        content: it only says "something changed". With a running ``events()``
+        loop this returns ``None``; otherwise it returns the polled batch, for
+        the caller to handle and ``ack``.
+        """
+        if not isinstance(verified, VerifiedWebhook):
+            from adcp.webhook_receiver import WebhookOutcome
+
+            if not isinstance(verified, WebhookOutcome):
+                raise TypeError(
+                    "handle_webhook takes a verified delivery: VerifiedWebhook.from_outcome("
+                    "await receiver.receive(...)). Verify the signature first."
+                )
+            verified = VerifiedWebhook.from_outcome(verified)
+        if self._listening and self._doorbell is not None:
+            self._doorbell.set()
+            return None
+        return await self.poll()
+
+
+def _merge_partial(
+    previous: Mapping[str, Any], current: dict[str, Any], report: ReconcileReport
+) -> dict[str, Any]:
+    unread = {issue.account_id for issue in report.issues if issue.account_id}
+    everything = not report.accounts or any(issue.account_id is None for issue in report.issues)
+    merged = dict(current)
+    for section in ("periods", "exceptions", "cutovers"):
+        values = dict(current.get(section) or {})
+        for key, value in (previous.get(section) or {}).items():
+            if everything or value.get("account_id") in unread:
+                values.setdefault(key, value)
+        merged[section] = values
+    if everything:
+        merged["covered"] = sorted(
+            set(current.get("covered") or ()) | set(previous.get("covered") or ())
+        )
+        merged["authorizations"] = {
+            **dict(previous.get("authorizations") or {}),
+            **dict(current.get("authorizations") or {}),
+        }
+    return merged
+
+
+__all__ = [
+    "EVENT_TYPES",
+    "SNAPSHOT_VERSION",
+    "EventBatch",
+    "InMemoryReportingEventStore",
+    "ReportingEvent",
+    "ReportingEventConflictError",
+    "ReportingEventStore",
+    "ReportingEvents",
+    "VerifiedWebhook",
+    "derive_events",
+]
diff --git a/src/adcp/reporting/subscriptions/facade.py b/src/adcp/reporting/subscriptions/facade.py
index 76714f24c..616f58c07 100644
--- a/src/adcp/reporting/subscriptions/facade.py
+++ b/src/adcp/reporting/subscriptions/facade.py
@@ -35,6 +35,14 @@
 from adcp.types import GetPrincipalRequest, SyncPrincipalRequest
 
 if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+    from adcp.reporting.subscriptions.events import (
+        EventBatch,
+        ReportingEvent,
+        ReportingEvents,
+        ReportingEventStore,
+    )
     from adcp.reporting.subscriptions.explain import CoverageTrace
     from adcp.reporting.subscriptions.reconcile import ReconcileReport
 
@@ -267,6 +275,7 @@ def __init__(
         *,
         max_attempts: int = _DEFAULT_ATTEMPTS,
         idempotency_key: Callable[[], str] | None = None,
+        event_store: ReportingEventStore | None = None,
     ) -> None:
         if type(max_attempts) is not int or not 1 <= max_attempts <= 10:
             raise ValueError("max_attempts must be an integer in 1..10")
@@ -276,6 +285,8 @@ def __init__(
         from adcp.reporting.subscriptions.reconcile import BuyerReportingState
 
         self._state = BuyerReportingState()
+        self._event_store = event_store
+        self._events: ReportingEvents | None = None
 
     # -- public API ---------------------------------------------------------------
 
@@ -473,6 +484,54 @@ async def explain(
             now=now,
         )
 
+    def _event_source(self) -> ReportingEvents:
+        if self._events is None:
+            from adcp.reporting.subscriptions.events import ReportingEvents
+
+            self._events = ReportingEvents(self._client, self._state, store=self._event_store)
+        return self._events
+
+    async def poll(
+        self,
+        *,
+        grace: timedelta = DEFAULT_GRACE,
+        baseline: bool = False,
+        now: datetime | None = None,
+    ) -> EventBatch:
+        """One batch of reporting events since the last ``ack``; serverless-safe.
+
+        ``batch = await client.reporting.poll(); await handle(batch.events);
+        await batch.ack()``. Events are derived by comparing this run's
+        ``reconcile`` and coverage with the snapshot saved at the last ack, so
+        a crash before ``ack`` redelivers the same events with the same ids.
+        Pass ``event_store=`` to :class:`ReportingClient` for a durable store.
+        """
+        return await self._event_source().poll(grace=grace, baseline=baseline, now=now)
+
+    def events(
+        self,
+        *,
+        poll_every: timedelta = timedelta(minutes=5),
+        stop: Any = None,
+        grace: timedelta = DEFAULT_GRACE,
+    ) -> AsyncIterator[ReportingEvent]:
+        """``async for event in client.reporting.events(): ...``, acked per batch.
+
+        Polls every ``poll_every``, or at once when :meth:`handle_webhook`
+        rings. ``stop`` is an ``asyncio.Event`` that ends the loop.
+        """
+        return self._event_source().events(poll_every=poll_every, stop=stop, grace=grace)
+
+    async def handle_webhook(self, verified: Any) -> EventBatch | None:
+        """Treat a verified seller webhook as a doorbell.
+
+        Pass ``VerifiedWebhook.from_outcome(await receiver.receive(...))``.
+        With a running :meth:`events` loop it wakes the loop and returns
+        ``None``; otherwise it polls now and returns the batch, which the
+        caller handles and acks. The body is never trusted as event content.
+        """
+        return await self._event_source().handle_webhook(verified)
+
     # -- planning -------------------------------------------------------------------
 
     async def _plan_subscribe(
diff --git a/src/adcp/reporting/subscriptions/reconcile.py b/src/adcp/reporting/subscriptions/reconcile.py
index e264f32ff..c17f25535 100644
--- a/src/adcp/reporting/subscriptions/reconcile.py
+++ b/src/adcp/reporting/subscriptions/reconcile.py
@@ -735,6 +735,30 @@ async def reconcile_subscriptions(
     now: datetime | None = None,
 ) -> ReconcileReport:
     """What every covered account is owed, and its status. See :meth:`ReportingClient.reconcile`."""
+    report, _context = await _reconcile(
+        client,
+        state=state,
+        since=since,
+        accounts=accounts,
+        subscription_id=subscription_id,
+        grace=grace,
+        include_not_owed=include_not_owed,
+        now=now,
+    )
+    return report
+
+
+async def _reconcile(
+    client: Any,
+    *,
+    state: BuyerReportingState | None,
+    since: Any,
+    accounts: Sequence[str] | Literal["covered"],
+    subscription_id: str | None,
+    grace: timedelta,
+    include_not_owed: bool,
+    now: datetime | 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")
@@ -816,7 +840,7 @@ async def reconcile_subscriptions(
         counts[line.status] += 1
         if line.status != "not_owed":
             counts["owed"] += 1
-    return ReconcileReport(
+    report = ReconcileReport(
         as_of=iso_z(instant),
         definitive=complete and not context.issues,
         summary=ReconcileSummary(**counts),
@@ -824,6 +848,7 @@ async def reconcile_subscriptions(
         issues=tuple(context.issues),
         accounts=tuple(record.account_id for record in records),
     )
+    return report, context
 
 
 def _not_owed_line(trace: CoverageTrace) -> PeriodLine:
diff --git a/tests/test_reporting_subscription_events.py b/tests/test_reporting_subscription_events.py
new file mode 100644
index 000000000..cc598bf0d
--- /dev/null
+++ b/tests/test_reporting_subscription_events.py
@@ -0,0 +1,137 @@
+"""Buyer reporting events: ``poll``, ``events`` and ``handle_webhook`` (AdCP 3.3).
+
+Events come from comparing durable snapshots of what the buyer reads, so a
+replay before ``ack`` repeats the same ids, and a webhook is only a doorbell.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import timedelta
+
+import pytest
+
+from adcp.reporting.subscriptions import (
+    ReportingEventConflictError,
+    VerifiedWebhook,
+)
+from adcp.reporting.subscriptions.facade import ReportingClient
+from adcp.webhook_receiver import WebhookOutcome
+from tests.reporting_inprocess import InProcessBuyer
+from tests.test_reporting_subscription_managed import ACCOUNT, START, _ctx, _World
+
+pytestmark = pytest.mark.asyncio
+
+
+def _types(batch) -> list[str]:
+    return sorted(event.type for event in batch.events)
+
+
+async def test_the_first_poll_reports_the_present_and_ack_moves_on() -> None:
+    world = _World()
+    await world.subscribe()
+    world.clock.set(START + timedelta(hours=3, minutes=30))
+    await world.service.run_worker()
+    reporting = ReportingClient(InProcessBuyer(world.platform, _ctx))
+    batch = await reporting.poll(now=world.clock())
+    assert _types(batch) == [
+        "account.covered",
+        "revision.available",
+        "revision.available",
+        "subscription.state_changed",
+    ]
+    revision = next(e for e in batch.events if e.type == "revision.available")
+    assert revision.account_id == ACCOUNT and revision.reporting_revision_id
+    assert revision.id.startswith("rse_") and revision.agent_id
+    # Not acked: the same events, with the same ids, come again.
+    again = await reporting.poll(now=world.clock())
+    assert [e.id for e in again.events] == [e.id for e in batch.events]
+    await again.ack()
+    after = await reporting.poll(now=world.clock())
+    assert after.events == []
+    # The stale batch can no longer be acknowledged.
+    with pytest.raises(ReportingEventConflictError):
+        await batch.ack()
+
+
+async def test_missing_obligations_then_revisions_are_events() -> None:
+    world = _World()
+    await world.subscribe()
+    reporting = ReportingClient(InProcessBuyer(world.platform, _ctx))
+    world.clock.set(START + timedelta(minutes=30))
+    await world.service.run_worker()
+    first = await reporting.poll(baseline=True, now=world.clock())
+    assert first.events == [] and first.report.definitive, first.report.issues
+    await first.ack()
+    late = await reporting.poll(now=START + timedelta(hours=3, minutes=30))
+    assert _types(late) == ["period.missing_obligation"]
+    assert late.events[0].occurred_at == "2026-11-01T03:00:00Z"
+    await late.ack()
+    world.clock.set(START + timedelta(hours=3, minutes=30))
+    await world.service.run_worker()
+    delivered = await reporting.poll(now=world.clock())
+    assert _types(delivered) == ["revision.available", "revision.available"]
+
+
+async def test_losing_and_regaining_authority_are_events() -> None:
+    world = _World()
+    await world.subscribe()
+    reporting = ReportingClient(InProcessBuyer(world.platform, _ctx))
+    world.clock.set(START + timedelta(minutes=30))
+    await (await reporting.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()
+    lost = await reporting.poll(now=world.clock())
+    assert _types(lost) == ["account.unauthorized"]
+    await lost.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 reporting.poll(now=world.clock())
+    assert "account.regranted" in _types(back)
+
+
+async def test_handle_webhook_needs_a_verified_delivery_and_polls_inline() -> None:
+    world = _World()
+    await world.subscribe()
+    reporting = ReportingClient(InProcessBuyer(world.platform, _ctx))
+    with pytest.raises(TypeError, match="verified"):
+        await reporting.handle_webhook({"notification_type": "principal.changed"})
+    with pytest.raises(ValueError, match="not verified"):
+        await reporting.handle_webhook(WebhookOutcome(rejected=True))
+    with pytest.raises(TypeError):
+        VerifiedWebhook({}, "seller")
+    outcome = WebhookOutcome(sender_identity="https://seller.example|kid", payload=None)
+    batch = await reporting.handle_webhook(outcome)
+    assert batch is not None and "subscription.state_changed" in _types(batch)
+
+
+async def test_the_events_loop_acks_each_batch_and_wakes_on_the_doorbell() -> None:
+    world = _World()
+    await world.subscribe()
+    reporting = ReportingClient(InProcessBuyer(world.platform, _ctx))
+    stop = asyncio.Event()
+    seen: list[str] = []
+
+    async def consume() -> None:
+        async for event in reporting.events(poll_every=timedelta(hours=1), stop=stop):
+            seen.append(event.type)
+
+    task = asyncio.create_task(consume())
+    for _ in range(200):
+        if seen:
+            break
+        await asyncio.sleep(0.05)
+    assert "subscription.state_changed" in seen
+    # A verified webhook wakes the running loop instead of polling inline.
+    ringing = await reporting.handle_webhook(
+        VerifiedWebhook.from_outcome(WebhookOutcome(sender_identity="seller", payload=None))
+    )
+    assert ringing is None
+    stop.set()
+    await asyncio.wait_for(task, timeout=30)
+    # The loop acknowledged what it delivered: only clock-driven period events remain.
+    remaining = await reporting.poll()
+    assert "subscription.state_changed" not in _types(remaining)

From 4c641410944bfd1d02fe82bc0e40ab5d82c06894 Mon Sep 17 00:00:00 2001
From: Brian O'Kelley <bokelley@scope3.com>
Date: Sun, 1 Nov 2026 17:39:41 +0000
Subject: [PATCH 2/5] fix(reporting): keep subscriptions and losses across a
 partial events poll

A poll that could not read the principal or the roster now keeps the
previous subscriptions and losses, so it never emits a state change or a
re-grant it did not observe.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 src/adcp/reporting/subscriptions/events.py  | 10 ++++++++++
 tests/test_reporting_subscription_events.py | 20 ++++++++++++++++++++
 2 files changed, 30 insertions(+)

diff --git a/src/adcp/reporting/subscriptions/events.py b/src/adcp/reporting/subscriptions/events.py
index 14ddd88ec..8b53c94ae 100644
--- a/src/adcp/reporting/subscriptions/events.py
+++ b/src/adcp/reporting/subscriptions/events.py
@@ -597,6 +597,16 @@ def _merge_partial(
                 values.setdefault(key, value)
         merged[section] = values
     if everything:
+        # A run that could not read the principal or the roster keeps the
+        # previous subscriptions and losses, so nothing reads as changed.
+        merged["subscriptions"] = {
+            **dict(previous.get("subscriptions") or {}),
+            **dict(current.get("subscriptions") or {}),
+        }
+        merged["unauthorized"] = {
+            **dict(previous.get("unauthorized") or {}),
+            **dict(current.get("unauthorized") or {}),
+        }
         merged["covered"] = sorted(
             set(current.get("covered") or ()) | set(previous.get("covered") or ())
         )
diff --git a/tests/test_reporting_subscription_events.py b/tests/test_reporting_subscription_events.py
index cc598bf0d..2f2b7ecaa 100644
--- a/tests/test_reporting_subscription_events.py
+++ b/tests/test_reporting_subscription_events.py
@@ -135,3 +135,23 @@ async def consume() -> None:
     # The loop acknowledged what it delivered: only clock-driven period events remain.
     remaining = await reporting.poll()
     assert "subscription.state_changed" not in _types(remaining)
+
+
+async def test_a_failed_read_never_fabricates_events() -> None:
+    world = _World()
+    await world.subscribe()
+    client = InProcessBuyer(world.platform, _ctx)
+    reporting = ReportingClient(client)
+    world.clock.set(START + timedelta(minutes=30))
+    await world.service.run_worker()
+    account = world.seller.accounts.pop(ACCOUNT)
+    await world.service.touch(account_id=ACCOUNT)
+    await world.service.drain_subscription_triggers()
+    await (await reporting.poll(now=world.clock())).ack()
+    for task in ("get_principal", "list_accounts"):
+        client.fail.add(task)
+        batch = await reporting.poll(now=world.clock())
+        assert batch.events == [], (task, [e.type for e in batch.events])
+        await batch.ack()
+        client.fail.discard(task)
+    world.seller.accounts[ACCOUNT] = account

From 25a6e7b0e201b8273831c3dc31944545bf35c389 Mon Sep 17 00:00:00 2001
From: Brian O'Kelley <bokelley@scope3.com>
Date: Sun, 1 Nov 2026 18:11:09 +0000
Subject: [PATCH 3/5] fix(reporting): consume the events doorbell before each
 poll

A webhook that rang during a poll, the loop body or the ack was cleared
after the cycle, so its change waited for the next timer poll. The ring is
now consumed before polling, so a ring during the cycle polls again at once.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 src/adcp/reporting/subscriptions/events.py  |  4 ++-
 tests/test_reporting_subscription_events.py | 34 +++++++++++++++++++++
 2 files changed, 37 insertions(+), 1 deletion(-)

diff --git a/src/adcp/reporting/subscriptions/events.py b/src/adcp/reporting/subscriptions/events.py
index 8b53c94ae..9f61dba51 100644
--- a/src/adcp/reporting/subscriptions/events.py
+++ b/src/adcp/reporting/subscriptions/events.py
@@ -541,11 +541,13 @@ async def events(
         self._listening += 1
         try:
             while stop is None or not stop.is_set():
+                # Consume the ring before polling: one that lands during the
+                # poll, the loop body or the ack survives and polls again at once.
+                self._doorbell.clear()
                 batch = await self.poll(grace=grace)
                 for event in batch.events:
                     yield event
                 await batch.ack()
-                self._doorbell.clear()
                 waiters = [asyncio.ensure_future(self._doorbell.wait())]
                 if stop is not None:
                     waiters.append(asyncio.ensure_future(stop.wait()))
diff --git a/tests/test_reporting_subscription_events.py b/tests/test_reporting_subscription_events.py
index 2f2b7ecaa..ce803cc9a 100644
--- a/tests/test_reporting_subscription_events.py
+++ b/tests/test_reporting_subscription_events.py
@@ -155,3 +155,37 @@ async def test_a_failed_read_never_fabricates_events() -> None:
         await batch.ack()
         client.fail.discard(task)
     world.seller.accounts[ACCOUNT] = account
+
+
+async def test_a_ring_during_a_cycle_is_not_swallowed() -> None:
+    world = _World()
+    await world.subscribe()
+    reporting = ReportingClient(InProcessBuyer(world.platform, _ctx))
+    source = reporting._event_source()
+    polls = 0
+    original = source.poll
+
+    async def counting(**kwargs):
+        nonlocal polls
+        polls += 1
+        batch = await original(**kwargs)
+        if polls == 1:
+            # A webhook rings while the first cycle is still running.
+            source._doorbell.set()
+        return batch
+
+    source.poll = counting
+    stop = asyncio.Event()
+
+    async def consume() -> None:
+        async for _event in reporting.events(poll_every=timedelta(hours=1), stop=stop):
+            pass
+
+    task = asyncio.create_task(consume())
+    for _ in range(200):
+        if polls >= 2:
+            break
+        await asyncio.sleep(0.05)
+    stop.set()
+    await asyncio.wait_for(task, timeout=30)
+    assert polls >= 2

From cf71c7dfad87e367ec36db0f83f7710ef1286389 Mon Sep 17 00:00:00 2001
From: Brian O'Kelley <bokelley@scope3.com>
Date: Sun, 1 Nov 2026 18:40:36 +0000
Subject: [PATCH 4/5] fix(reporting): keep the events loop running after an ack
 conflict

A ReportingEventConflictError is transient: another worker acknowledged
the snapshot first. events now polls again instead of ending.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 src/adcp/reporting/subscriptions/events.py  |  7 ++++-
 tests/test_reporting_subscription_events.py | 31 +++++++++++++++++++++
 2 files changed, 37 insertions(+), 1 deletion(-)

diff --git a/src/adcp/reporting/subscriptions/events.py b/src/adcp/reporting/subscriptions/events.py
index 9f61dba51..1cd82fe1b 100644
--- a/src/adcp/reporting/subscriptions/events.py
+++ b/src/adcp/reporting/subscriptions/events.py
@@ -547,7 +547,12 @@ async def events(
                 batch = await self.poll(grace=grace)
                 for event in batch.events:
                     yield event
-                await batch.ack()
+                try:
+                    await batch.ack()
+                except ReportingEventConflictError:
+                    # Another worker acknowledged first: poll again at once.
+                    # Events it already acknowledged are not redelivered.
+                    continue
                 waiters = [asyncio.ensure_future(self._doorbell.wait())]
                 if stop is not None:
                     waiters.append(asyncio.ensure_future(stop.wait()))
diff --git a/tests/test_reporting_subscription_events.py b/tests/test_reporting_subscription_events.py
index ce803cc9a..03147df71 100644
--- a/tests/test_reporting_subscription_events.py
+++ b/tests/test_reporting_subscription_events.py
@@ -189,3 +189,34 @@ async def consume() -> None:
     stop.set()
     await asyncio.wait_for(task, timeout=30)
     assert polls >= 2
+
+
+async def test_the_events_loop_survives_an_acknowledgement_conflict() -> None:
+    world = _World()
+    await world.subscribe()
+    reporting = ReportingClient(InProcessBuyer(world.platform, _ctx))
+    other = ReportingClient(InProcessBuyer(world.platform, _ctx), event_store=None)
+    source = reporting._event_source()
+    other._event_source().store = source.store  # two workers, one store
+    stop = asyncio.Event()
+    seen: list[str] = []
+    raced = False
+
+    async def consume() -> None:
+        nonlocal raced
+        async for event in reporting.events(poll_every=timedelta(hours=1), stop=stop):
+            seen.append(event.type)
+            if not raced:
+                raced = True
+                # The other worker acknowledges the same snapshot first.
+                await (await other.poll()).ack()
+
+    task = asyncio.create_task(consume())
+    for _ in range(200):
+        if raced and task.done() is False and len(seen) >= 1:
+            break
+        await asyncio.sleep(0.05)
+    await asyncio.sleep(0.5)
+    assert not task.done(), task.exception() if task.done() else None
+    stop.set()
+    await asyncio.wait_for(task, timeout=30)

From db016f898fbd49d9e33b2e8b3fc67335c6c726ad Mon Sep 17 00:00:00 2001
From: Brian O'Kelley <bokelley@scope3.com>
Date: Sun, 1 Nov 2026 19:08:48 +0000
Subject: [PATCH 5/5] fix(reporting): give each account rejoin a new
 account.covered event id

An account that left a subscription and rejoined reused its first
account.covered id, so consumers deduplicating on event.id dropped the
rejoin. Each join now carries an occurrence number in the snapshot and the
id.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 src/adcp/reporting/subscriptions/events.py  |  8 +++++++-
 tests/test_reporting_subscription_events.py | 17 +++++++++++++++++
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/src/adcp/reporting/subscriptions/events.py b/src/adcp/reporting/subscriptions/events.py
index 1cd82fe1b..0421fa79b 100644
--- a/src/adcp/reporting/subscriptions/events.py
+++ b/src/adcp/reporting/subscriptions/events.py
@@ -361,16 +361,22 @@ def emit(
             )
 
     old_covered = set(before.get("covered") or ())
+    # Each time an account (re)joins a subscription is a new occurrence, so a
+    # rejoin is never deduplicated against the first join.
+    generations = dict(before.get("covered_generations") or {})
     for item in current.get("covered") or ():
         if item not in old_covered:
+            generations[item] = int(generations.get(item, 0)) + 1
             subscription_id, account_id = json.loads(item)
             emit(
                 "account.covered",
                 item,
-                1,
+                generations[item],
                 None,
                 {"subscription_id": subscription_id, "account_id": account_id},
             )
+    if isinstance(current, dict):
+        current["covered_generations"] = generations
 
     old_unauthorized = before.get("unauthorized") or {}
     new_unauthorized = current.get("unauthorized") or {}
diff --git a/tests/test_reporting_subscription_events.py b/tests/test_reporting_subscription_events.py
index 03147df71..e6d117977 100644
--- a/tests/test_reporting_subscription_events.py
+++ b/tests/test_reporting_subscription_events.py
@@ -220,3 +220,20 @@ async def consume() -> None:
     assert not task.done(), task.exception() if task.done() else None
     stop.set()
     await asyncio.wait_for(task, timeout=30)
+
+
+def test_a_rejoin_is_a_new_account_covered_event() -> None:
+    from datetime import datetime, timezone
+
+    from adcp.reporting.subscriptions.events import derive_events
+
+    now = datetime(2026, 11, 1, tzinfo=timezone.utc)
+    key = '["analytics", "acc_a"]'
+    joined = {"covered": [key]}
+    first = derive_events({}, joined, agent_id="a", observed_at=now)
+    left = {"covered": [], "covered_generations": joined["covered_generations"]}
+    assert derive_events(joined, left, agent_id="a", observed_at=now) == []
+    rejoined = {"covered": [key]}
+    again = derive_events(left, rejoined, agent_id="a", observed_at=now)
+    assert [e.type for e in first] == [e.type for e in again] == ["account.covered"]
+    assert first[0].id != again[0].id