feat(reporting): add the reporting test kit, reference seller and pytest fixture

README.md

@@ -375,7 +375,7 @@ outcome = await client.accounts.ensure(key, billing="operator")  # AccountReady
 - **Explain.** Seller `reporting.explain_coverage(...)` and buyer `client.reporting.explain(...)` return the same `CoverageTrace`; `python -m adcp reporting diff` compares them.
 - **Events.** `client.reporting.poll()` / `events()` derive events from saved state with deterministic ids; `handle_webhook()` treats a verified webhook as a doorbell.
 - **Receipts.** Off by default; `reconcile(receipts=ReceiptPolicy(mode="verified", commit=...))` receipts a billing revision only after the SDK verified it and your commit returned a reference.
-- **Coming in later betas:** the test kit.
+- **Test kit.** `adcp.testing.reporting.reporting_test_kit(...)` (and the `adcp_reporting_kit` pytest fixture) runs the reference seller under a virtual clock with real buyers and `assert_agreement`.
 
 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-testing.md

@@ -2,11 +2,59 @@
 
 ## Testing today
 
-`ReliableReportingService(storage=memory_storage(), subscriptions=...)` runs in process under pytest. Drive it with a `DeterministicReportingClock` and `run_worker()`, and call `await reporting.drain_subscription_triggers()` before asserting on triggered reconciles. `tests/test_reporting_subscription_managed.py` is a worked example.
+`reporting_test_kit(...)` runs a seller in process under a virtual clock and gives you real buyer clients bound to it. The SDK's reference seller is the default. Pass your own platform to test your wiring.
+
+```python
+from adcp.testing.reporting import offerings, reporting_test_kit
+
+async with reporting_test_kit(
+    start="2026-11-01T00:00:00-04:00",
+    offerings=[offerings.daily_analytics(alignment="account_timezone")],
+    accounts=[{"account_id": "acc_a", "timezone": "America/New_York"}],
+    buys=[{"media_buy_id": "mb_1", "account_id": "acc_a", "confirmed_at": "2026-11-03T15:00:00Z",
+           "start_time": "2026-11-03T05:00:00Z", "end_time": "2026-11-30T05:00:00Z"}],
+) as kit:
+    buyer = kit.buyer("agency-x")                       # a real client.reporting
+    await buyer.reporting.subscribe(analytics_daily)
+    await kit.clock.advance("P3D")                      # the worker runs at each boundary crossed
+    await kit.grants.revoke("agency-x", "acc_a")
+    await kit.clock.advance("P1D")
+    await kit.grants.regrant("agency-x", "acc_a")
+    await kit.grants.wind_down("agency-x", "acc_a")
+    await kit.accounts.set_timezone("acc_a", "America/Los_Angeles", principal="agency-x")
+    await kit.clock.advance_to("2027-03-14T12:00:00-04:00")   # DST
+    await kit.buys.cancel("mb_1", principal="agency-x")
+    await kit.assert_agreement("agency-x")
+```
+
+- **Clock.** `kit.clock.advance("P3D")` (days to seconds) and `advance_to(instant)` move the one virtual clock forward. At every period boundary crossed, the seller processes the move as production would: the boundary sweep, then planning.
+- **Grants, accounts and buys.** `kit.grants` (`revoke`, `regrant`, `reduce`, `wind_down`), `kit.accounts.set_timezone` and `kit.buys` (`add`, `cancel`) go through the controller's `reporting_subscription_probe` and `seed_media_buy`. That is the same scenario a storyboard drives against a remote seller.
+- **Agreement.** `await kit.assert_agreement(principal)` checks that the seller's owed set equals what the buyer's `reconcile()` computes over the wire. It also checks that the `authorized_at` served on `list_accounts` is the one coverage used. On a difference, the `AgreementError` names each disputed period, the first rule at which the two `CoverageTrace`s differ, and both sentences.
+- **Your platform.** Pass `platform=` (your installed handler), `service=` (your `ReliableReportingService`) and `controller=` (a `TestControllerStore` serving `reporting_subscription_probe` and `seed_media_buy`). The kit then drives your seller through that controller.
+- **Offerings.** `offerings.daily_analytics(...)` and `offerings.hourly_pacing(...)` build offering fixtures that publish `standard_dimensions`.
 
 ## The test kit
 
-`reporting_test_kit(...)` and the `adcp_reporting_kit` pytest fixture arrive in a later beta. They add a virtual clock, grants, accounts, buys, and `assert_agreement`. Python sellers run the TypeScript storyboard CLI against their URL:
+The pytest fixture `adcp_reporting_kit` yields a started kit and closes it after the test:
+
+```python
+# conftest.py
+pytest_plugins = ["adcp.testing.reporting_pytest"]
+
+# test_reporting.py
+@pytest.mark.adcp_reporting_kit(accounts=[{"account_id": "acc_a", "timezone": "UTC"}])
+async def test_owed(adcp_reporting_kit):
+    buyer = adcp_reporting_kit.buyer("agency-x")
+    await buyer.reporting.subscribe(analytics_daily)
+    await adcp_reporting_kit.clock.advance("P3D")
+    await adcp_reporting_kit.assert_agreement("agency-x")
+```
+
+The marker's keyword arguments are those of `reporting_test_kit`. Run the fixture under pytest-asyncio.
+
+For a lower-level test, `ReliableReportingService(storage=memory_storage(), subscriptions=...)` runs in process under pytest. Drive it with a `DeterministicReportingClock` and `run_worker()`, and call `await reporting.drain_subscription_triggers()` before asserting on triggered reconciles. `tests/test_reporting_subscription_managed.py` is a worked example.
+
+Python sellers run the TypeScript storyboard CLI against their URL:
 
 ```
 npx adcp storyboard run https://seller.example/mcp --storyboards reporting_subscriptions --auth "$TOKEN"

src/adcp/reporting/subscriptions/engine.py

@@ -65,6 +65,7 @@
     SubscriptionBuy,
     SubscriptionException,
     SubscriptionVersionWindow,
+    _timezone_at,
     authorization_chain_start,
     authorization_regranted_at,
     owed_subscription_periods,
@@ -1892,7 +1893,11 @@ def _schedules(
         destinations: Mapping[str, Any],
         now: datetime,
     ) -> list[tuple[str, PeriodCalendar]]:
-        """``(offering_id, calendar)`` for each offering the account resolves to."""
+        """``(offering_id, calendar)`` for each offering the account resolves to at ``now``.
+
+        Each calendar is in the zone in effect at ``now``, so a gap served for
+        a loss before a timezone change stays on the calendar it removed.
+        """
         if account is None:
             return []
         selector = SubscriptionSelector.from_subscription(state["configuration"])
@@ -1912,7 +1917,13 @@ def _schedules(
             ).offerings:
                 found[item.offering_id] = item
         return [
-            (item.offering_id, item.calendar(account.timezone, selector.period_timezone))
+            (
+                item.offering_id,
+                item.calendar(
+                    _timezone_at(now, account.timezone, account.timezone_changes, item.offering_id),
+                    selector.period_timezone,
+                ),
+            )
             for item in sorted(found.values(), key=lambda value: value.offering_id.encode())
         ]
 
@@ -2112,7 +2123,10 @@ def _gaps(
                 continue
             if expired:
                 continue  # a fresh coverage start claims no gap
-            schedules = self._schedules(state, account, buys, destinations, lost)
+            # The gap removed periods of the schedules in force at the loss:
+            # those of the version then current, in the zone then in effect,
+            # never a later version's (adcp-client-python#1770).
+            schedules = self._schedules(_state_at(state, lost), account, buys, destinations, lost)
             entries = [
                 (offering_id, calendar.floor(lost), calendar.ceil(regranted))
                 for offering_id, calendar in schedules
@@ -2483,6 +2497,21 @@ async def receipt_roster(self, identity: PrincipalLike, account_id: str) -> list
 # -- helpers -------------------------------------------------------------------
 
 
+def _state_at(state: Mapping[str, Any], instant: datetime) -> Mapping[str, Any]:
+    """``state`` with the configuration of the version in force at ``instant``."""
+    for prior in state.get("prior_versions") or ():
+        accepted = _instant(prior.get("accepted_at"))
+        superseded = _instant(prior.get("superseded_at"))
+        if (
+            prior.get("configuration") is not None
+            and accepted is not None
+            and accepted <= instant
+            and (superseded is None or instant < superseded)
+        ):
+            return {**state, "configuration": prior["configuration"]}
+    return state
+
+
 def _buy_offering_ids(buys: Sequence[SubscriptionBuy]) -> list[str]:
     ids: set[str] = set()
     for buy in buys:

src/adcp/testing/decisioning.py

@@ -289,8 +289,7 @@ def build_asgi_app(
         raise ValueError(f"Unsupported transport {transport!r}; expected 'mcp', 'a2a', or 'both'.")
     if (on_startup or on_shutdown) and transport != "both":
         raise ValueError(
-            "on_startup / on_shutdown hooks require transport='both', "
-            f"got transport={transport!r}."
+            f"on_startup / on_shutdown hooks require transport='both', got transport={transport!r}."
         )
     if transport == "a2a":
         ignored_session_settings = []