fix(reporting): read each package offering snapshot from get_media_buys

docs/reliable-reporting-service.md

@@ -345,7 +345,7 @@ platform = reporting.install(MySeller())  # then serve the platform as usual
 | The principal runtime, `get_principal`/`sync_principal` and the principal section of capabilities | the service's own `PrincipalService` | `principals`, `principal_store` |
 | The accounts a principal may read | the platform's `list_accounts` (a decisioning platform's `accounts.list`), read under the principal's identity | `accounts(principal, *, cursor)` |
 | Each account's buys | the platform's `get_media_buys`, under the same identity | `buys(principal, account_id)` |
-| Each package's offerings | every registered offering | `buy_offerings(media_buy)` |
+| Each package's offerings | the package's `reporting_delivery_offering_ids` snapshot from `get_media_buys` (`[]` means none), else every registered offering | `buy_offerings(media_buy)` |
 | Offerings | the registered `capability_offerings`, each publishing `standard_dimensions` | none |
 | Report content per generation | `account_context` | `configure(generation)` |
 | The authenticated principal | the caller the reporting tasks resolve, with `ToolContext.tenant_id` | `resolve_principal(request, context)` |

docs/reporting-subscriptions-seller.md

@@ -25,7 +25,7 @@ The SDK reads these surfaces under each principal's identity. A principal sees e
 |---|---|---|---|
 | Authorized accounts | `list_accounts` (a decisioning platform's `accounts.list`) | `account_id`, `timezone` (IANA), `operator`, `authorization.allowed_tasks` | `SubscriptionOptions.accounts` |
 | Visible buys | `get_media_buys` | `confirmed_at`, `start_time`, `end_time`, `cancellation.canceled_at`, `cancellation_effective_at`, `completed_at`, `flight_history`, `packages[].package_id` | `SubscriptionOptions.buys` |
-| Package offerings | `packages[].reporting_delivery_offering_ids` (fixed at confirmation), else every registered offering | | `SubscriptionOptions.buy_offerings` |
+| Package offerings | `packages[].reporting_delivery_offering_ids` (fixed at confirmation; `[]` means none), else, only when the seller omits it, every registered offering | | `SubscriptionOptions.buy_offerings` |
 | Offerings | `capability_offerings` on each registered source | `standard_dimensions` on every offering; `available_from` / `withdrawn_at` when they apply; `account_id` in `reporting_profile.primary_keys` on every offering delivered to destinations (`method`), required by 3.3.0-beta.3 and logged as `reporting.subscription.offering_primary_keys` when missing | none |
 | Report content | `account_context`, called with `configuration.subscription` set | | `SubscriptionOptions.configure` |
 

src/adcp/reporting/subscriptions/managed.py

@@ -104,6 +104,19 @@
 _MAX_PAGES = 10_000
 
 
+def _package_offering_ids(package: Mapping[str, Any], every: Sequence[str]) -> list[str]:
+    """A package's ``reporting_delivery_offering_ids`` snapshot (3.3.0-beta.3).
+
+    The seller fixes it when the package is confirmed and serves ``[]`` when
+    no offering applies, so an empty snapshot is honoured. Only a package
+    served without one falls back to every registered offering.
+    """
+    snapshot = package.get("reporting_delivery_offering_ids")
+    if isinstance(snapshot, (list, tuple)):
+        return [str(item) for item in snapshot]
+    return list(every)
+
+
 @dataclass(frozen=True)
 class SubscriptionOptions:
     """Serve ``media_buy.reporting_subscriptions`` from ``ReliableReportingService``.
@@ -121,8 +134,11 @@ class SubscriptionOptions:
     * ``buys(principal, account_id)``: an account's buys, instead of
       ``get_media_buys``.
     * ``buy_offerings(media_buy)``: each package's offering ids, keyed by
-      ``package_id``. Without it every package is reported by every offering
-      the service registered; supply it when products differ.
+      ``package_id``. Without it each package's own
+      ``reporting_delivery_offering_ids`` snapshot from ``get_media_buys`` is
+      used (3.3.0-beta.3, fixed at confirmation; ``[]`` means no offering), and
+      only a package without one is reported by every offering the service
+      registered.
     * ``configure(generation)``: the report content for a subscription
       generation, instead of the service's ``account_context``.
     * ``account_delivery_configs(principal, account_id)``: the caller's own
@@ -986,8 +1002,10 @@ def _buy(self, raw: Mapping[str, Any]) -> SubscriptionBuy | None:
 
         The flight, ``cancellation_effective_at`` and ``flight_history`` come
         from the wire. Each package's offering set is ``buy_offerings`` when
-        given, else the package's ``reporting_delivery_offering_ids`` (fixed at
-        confirmation), else every registered offering.
+        given, else the package's ``reporting_delivery_offering_ids`` snapshot
+        (fixed at confirmation; an empty one means no offering reports the
+        package), else, only when the seller omits the snapshot, every
+        registered offering.
         """
         if raw.get("start_time") is None or raw.get("end_time") is None:
             return None  # not yet flighted: nothing can be owed for it
@@ -1003,9 +1021,7 @@ def _buy(self, raw: Mapping[str, Any]) -> SubscriptionBuy | None:
             media_buy["packages"] = [
                 {
                     **package,
-                    "reporting_delivery_offering_ids": list(
-                        package.get("reporting_delivery_offering_ids") or every
-                    ),
+                    "reporting_delivery_offering_ids": _package_offering_ids(package, every),
                 }
                 for package in raw.get("packages") or ()
                 if package.get("package_id")

tests/conformance/reporting/test_reporting_row_storage_object_pg.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
@@ -442,13 +443,18 @@ async def test_the_producer_stages_object_uploads_outside_its_publication_transa
     from tests.test_reporting_settling import _capabilities, _harness, _revisions
 
     async with isolated_reporting_pool() as pool:
-        producer, store, _, _ = await _harness(
+        producer, store, _, clock = await _harness(
             _capabilities(restatement_window="P3D"),
             store_factory=lambda clock: PgReportingLedgerStore(
                 pool=pool, clock=clock, row_storage=options(tmp_path)
             ),
         )
         assert isinstance(producer, ReportingProducer)
+        # The row upload checks the worker lease against the database's
+        # clock_timestamp(), so the lease the injected clock stamps must lie in
+        # the real future: run the turn at the wall clock, not the harness's
+        # fixed instant, which is already in the past.
+        clock[0] = datetime.now(timezone.utc)
         turn = await producer.run_worker()
         assert turn.revisions_committed
         (revision,) = await _revisions(store)

tests/test_reporting_subscription_managed.py

@@ -322,6 +322,39 @@ async def test_derived_buys_use_buy_offerings_per_package() -> None:
     assert turn.subscription_obligations_written == 0
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("snapshot", "owed"),
+    [
+        # The snapshot names the registered offering: the package is reported.
+        ([OFFERING["offering_id"]], True),
+        # 3.3.0-beta.3: [] means no offering applies, never "every offering".
+        ([], False),
+        # The snapshot fixed at confirmation names another offering.
+        (["SOME_OTHER_OFFERING"], False),
+    ],
+)
+async def test_derived_buys_read_each_package_offering_snapshot(
+    snapshot: list[str], owed: bool
+) -> None:
+    world = _World()
+    world.seller.buys[0]["packages"][0]["reporting_delivery_offering_ids"] = snapshot
+    await world.subscribe()
+    world.clock.set(START + timedelta(hours=3, minutes=30))
+    turn = await world.service.run_worker()
+    assert (turn.subscription_obligations_written > 0) is owed
+
+
+@pytest.mark.asyncio
+async def test_a_package_without_a_snapshot_falls_back_to_every_registered_offering() -> None:
+    world = _World()
+    assert "reporting_delivery_offering_ids" not in world.seller.buys[0]["packages"][0]
+    await world.subscribe()
+    world.clock.set(START + timedelta(hours=3, minutes=30))
+    turn = await world.service.run_worker()
+    assert turn.subscription_obligations_written > 0
+
+
 @pytest.mark.asyncio
 async def test_replace_account_configs_is_refused_without_a_host_that_applies_it() -> None:
     world = _World()